feat: add metrics for auth_query and SecretChecker erpc calls

lib/supavisor/application.ex

@@ -131,6 +131,7 @@ defmodule Supavisor.Application do
           child_spec: DynamicSupervisor, strategy: :one_for_one, name: Supavisor.DynamicSupervisor
         },
         Supavisor.Vault,
+        Supavisor.ConnectionListener,
 
         # Start the Endpoint (http/https)
         SupavisorWeb.Endpoint

lib/supavisor/auth_query.ex

@@ -6,6 +6,7 @@ defmodule Supavisor.AuthQuery do
 
   alias Supavisor.Errors.AuthQueryError
   alias Supavisor.Helpers
+  alias Supavisor.Monitoring.Telem
   alias Supavisor.Secrets.{ManagerSecrets, SASLSecrets}
   alias Supavisor.Tenants.Tenant
 
@@ -17,27 +18,33 @@ defmodule Supavisor.AuthQuery do
   def start_link(%Tenant{} = tenant, %ManagerSecrets{} = manager) do
     ssl_opts = build_ssl_options(tenant)
     ip_version = Helpers.ip_version(tenant.ip_version, tenant.db_host)
+    start = System.monotonic_time()
 
-    case Postgrex.start_link(
-           hostname: tenant.db_host,
-           port: tenant.db_port,
-           database: tenant.db_database,
-           password: manager.db_password,
-           username: manager.db_user,
-           parameters: [application_name: "Supavisor (auth_query)"],
-           ssl: tenant.upstream_ssl,
-           socket_options: [ip_version],
-           queue_target: 1_000,
-           queue_interval: 5_000,
-           ssl_opts: ssl_opts
-         ) do
+    [
+      hostname: tenant.db_host,
+      port: tenant.db_port,
+      database: tenant.db_database,
+      password: manager.db_password,
+      username: manager.db_user,
+      parameters: [application_name: "Supavisor (auth_query)"],
+      ssl: tenant.upstream_ssl,
+      socket_options: [ip_version],
+      queue_target: 1_000,
+      queue_interval: 5_000,
+      ssl_opts: ssl_opts,
+      connection_listeners: {[Supavisor.ConnectionListener], start}
+    ]
+    |> Postgrex.start_link()
+    |> case do
       {:ok, pid} ->
         {:ok, pid}
 
       :ignore ->
+        telemetry_connection_stop(start, :connection_failed)
         {:error, %AuthQueryError{reason: :connection_failed, details: "connection ignored"}}
 
       {:error, reason} ->
+        telemetry_connection_stop(start, :connection_failed)
         {:error, %AuthQueryError{reason: :connection_failed, details: inspect(reason)}}
     end
   end
@@ -96,11 +103,19 @@ defmodule Supavisor.AuthQuery do
   """
   @spec fetch_user_secret(pid(), String.t() | nil, String.t()) ::
           {:ok, SASLSecrets.t()} | {:error, AuthQueryError.t()}
-  def fetch_user_secret(_conn, nil, _user) do
+  def fetch_user_secret(conn, auth_query, user) do
+    start = System.monotonic_time()
+
+    conn
+    |> do_fetch_user_secret(auth_query, user)
+    |> tap(&telemetry_query_stop(start, &1))
+  end
+
+  defp do_fetch_user_secret(_conn, nil, _user) do
     {:error, %AuthQueryError{reason: :no_auth_query}}
   end
 
-  def fetch_user_secret(conn, auth_query, user) when is_binary(auth_query) do
+  defp do_fetch_user_secret(conn, auth_query, user) when is_binary(auth_query) do
     Postgrex.query!(conn, auth_query, [user])
   catch
     _error, reason ->
@@ -167,4 +182,13 @@ defmodule Supavisor.AuthQuery do
   defp build_ssl_options(_tenant) do
     [verify: :verify_none]
   end
+
+  defp telemetry_connection_stop(start, error),
+    do: Telem.auth_query_connection_stop(System.monotonic_time() - start, telemetry_status(error))
+
+  defp telemetry_query_stop(start, result),
+    do: Telem.auth_query_query_stop(System.monotonic_time() - start, telemetry_status(result))
+
+  defp telemetry_status({:ok, _}), do: :ok
+  defp telemetry_status(_), do: :error
 end

lib/supavisor/client_handler.ex

@@ -1009,6 +1009,7 @@ defmodule Supavisor.ClientHandler do
 
     error = %ClientSocketClosedError{mode: data.mode, client_state: state}
     context = if state in [:idle, :busy], do: :authenticated, else: :handshake
+
     Error.terminate_with_error(data, error, context)
   end
 end

lib/supavisor/connection_listener.ex

@@ -0,0 +1,88 @@
+defmodule Supavisor.ConnectionListener do
+  use GenServer
+
+  require Logger
+
+  @name __MODULE__
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, nil, Keyword.put_new(opts, :name, @name))
+  end
+
+  @impl true
+  def init(nil) do
+    Process.flag(:trap_exit, true)
+    {:ok, %{monitoring: %{}}}
+  end
+
+  @impl true
+  def handle_info({:connected, pid, connect_start_time}, state) do
+    {start_time, state} =
+      case Map.get(state.monitoring, pid) do
+        nil ->
+          ref = Process.monitor(pid)
+          {connect_start_time, put_in(state, [:monitoring, pid], {ref, nil})}
+
+        {ref, disconnect_time} ->
+          {disconnect_time || connect_start_time, put_in(state, [:monitoring, pid], {ref, nil})}
+      end
+
+    connect_duration = System.monotonic_time() - start_time
+
+    :telemetry.execute(
+      [:supavisor, :auth_query, :connection, :stop],
+      %{duration: connect_duration}
+    )
+
+    {:noreply, state}
+  end
+
+  def handle_info({:disconnected, pid, _}, state) do
+    case Map.get(state.monitoring, pid) do
+      nil ->
+        {:noreply, state}
+
+      {_ref, prev_disconnect_ts} when not is_nil(prev_disconnect_ts) ->
+        Logger.warning("Duplicate disconnected event for pid #{inspect(pid)})")
+        {:noreply, state}
+
+      {ref, nil} ->
+        :telemetry.execute(
+          [:supavisor, :auth_query, :disconnection],
+          %{count: 1}
+        )
+
+        {:noreply, put_in(state, [:monitoring, pid], {ref, System.monotonic_time()})}
+    end
+  end
+
+  def handle_info({:disconnected, pid}, state) do
+    handle_info({:disconnected, pid, nil}, state)
+  end
+
+  # we only get monitor message if we have seen the connected event first
+  def handle_info({:DOWN, _ref, :process, pid, reason}, state) do
+    case Map.pop(state.monitoring, pid) do
+      {nil, _} ->
+        {:noreply, state}
+
+      {{_ref, prev_disconnect_ts}, monitoring} when not is_nil(prev_disconnect_ts) ->
+        Logger.warning(
+          "Duplicate disconnected event for pid #{inspect(pid)} (#{inspect(reason)})"
+        )
+
+        {:noreply, %{state | monitoring: monitoring}}
+
+      {{ref, nil}, monitoring} ->
+        Process.demonitor(ref, [:flush])
+
+        :telemetry.execute(
+          [:supavisor, :auth_query, :disconnection],
+          %{count: 1},
+          (reason == :normal && %{}) || %{kind: :exit, reason: reason}
+        )
+
+        {:noreply, %{state | monitoring: monitoring}}
+    end
+  end
+end

lib/supavisor/monitoring/telem.ex

@@ -5,6 +5,7 @@ defmodule Supavisor.Monitoring.Telem do
   require Supavisor
 
   @disabled Application.compile_env(:supavisor, :metrics_disabled, false)
+  @slow_auth_query_ms 1_000
 
   if @disabled do
     defp telemetry_execute(_name, _measurements, _meta), do: :ok
@@ -117,6 +118,50 @@ defmodule Supavisor.Monitoring.Telem do
     )
   end
 
+  @spec auth_query_connection_stop(integer(), term()) :: :ok
+  def auth_query_connection_stop(duration, error_or_nil) do
+    telemetry_execute(
+      [:supavisor, :auth_query, :connection, :stop],
+      %{duration: duration},
+      %{error: error_or_nil}
+    )
+  end
+
+  @spec auth_query_query_stop(integer(), :ok | :error) :: :ok
+  def auth_query_query_stop(duration, status) do
+    duration_ms = System.convert_time_unit(duration, :native, :millisecond)
+
+    if duration_ms > @slow_auth_query_ms do
+      Logger.warning("auth_query took over #{@slow_auth_query_ms}ms (#{duration_ms}ms)")
+    end
+
+    telemetry_execute(
+      [:supavisor, :auth_query, :query, :stop],
+      %{duration: duration},
+      %{status: status}
+    )
+  end
+
+  @spec secret_checker_get_secrets_stop(integer(), :ok | :error, :local | :remote) :: :ok
+  def secret_checker_get_secrets_stop(duration, status, locality)
+      when locality in [:local, :remote] do
+    telemetry_execute(
+      [:supavisor, :secret_checker, :get_secrets, :stop, locality],
+      %{duration: duration},
+      %{status: status}
+    )
+  end
+
+  @spec secret_checker_update_credentials_stop(integer(), :ok | :error, :local | :remote) :: :ok
+  def secret_checker_update_credentials_stop(duration, status, locality)
+      when locality in [:local, :remote] do
+    telemetry_execute(
+      [:supavisor, :secret_checker, :update_credentials, :stop, locality],
+      %{duration: duration},
+      %{status: status}
+    )
+  end
+
   @spec id_to_tags(Supavisor.id()) :: map()
   defp id_to_tags(
          Supavisor.id(

lib/supavisor/monitoring/tenant.ex

@@ -27,7 +27,8 @@ defmodule Supavisor.PromEx.Plugins.Tenant do
     [
       system_metrics(),
       client_metrics(),
-      db_metrics()
+      db_metrics(),
+      client_auth_metrics()
     ]
   end
 
@@ -191,6 +192,68 @@ defmodule Supavisor.PromEx.Plugins.Tenant do
     )
   end
 
+  defp client_auth_metrics do
+    Event.build(
+      :supavisor_client_auth_event_metrics,
+      [
+        distribution(
+          [:supavisor, :auth_query, :connection, :duration],
+          event_name: [:supavisor, :auth_query, :connection, :stop],
+          measurement: :duration,
+          description: "Duration of auth_query Postgrex connection setup.",
+          tags: [:status],
+          unit: {:native, :millisecond},
+          reporter_options: [peep_bucket_calculator: Buckets]
+        ),
+        distribution(
+          [:supavisor, :auth_query, :query, :duration],
+          event_name: [:supavisor, :auth_query, :query, :stop],
+          measurement: :duration,
+          description: "Duration of auth_query execution including Postgrex queue wait.",
+          tags: [:status],
+          unit: {:native, :millisecond},
+          reporter_options: [peep_bucket_calculator: Buckets]
+        ),
+        distribution(
+          [:supavisor, :secret_checker, :get_secrets, :duration, :local],
+          event_name: [:supavisor, :secret_checker, :get_secrets, :stop, :local],
+          measurement: :duration,
+          description: "Duration of get_secrets erpc call (same-node).",
+          tags: [:status],
+          unit: {:native, :millisecond},
+          reporter_options: [peep_bucket_calculator: Buckets]
+        ),
+        distribution(
+          [:supavisor, :secret_checker, :get_secrets, :duration, :remote],
+          event_name: [:supavisor, :secret_checker, :get_secrets, :stop, :remote],
+          measurement: :duration,
+          description: "Duration of get_secrets erpc call (cross-node).",
+          tags: [:status],
+          unit: {:native, :millisecond},
+          reporter_options: [peep_bucket_calculator: Buckets]
+        ),
+        distribution(
+          [:supavisor, :secret_checker, :update_credentials, :duration, :local],
+          event_name: [:supavisor, :secret_checker, :update_credentials, :stop, :local],
+          measurement: :duration,
+          description: "Duration of update_credentials erpc call (same-node).",
+          tags: [:status],
+          unit: {:native, :millisecond},
+          reporter_options: [peep_bucket_calculator: Buckets]
+        ),
+        distribution(
+          [:supavisor, :secret_checker, :update_credentials, :duration, :remote],
+          event_name: [:supavisor, :secret_checker, :update_credentials, :stop, :remote],
+          measurement: :duration,
+          description: "Duration of update_credentials erpc call (cross-node).",
+          tags: [:status],
+          unit: {:native, :millisecond},
+          reporter_options: [peep_bucket_calculator: Buckets]
+        )
+      ]
+    )
+  end
+
   defp db_metrics do
     Event.build(
       :supavisor_tenant_db_event_metrics,

lib/supavisor/secret_checker.ex

@@ -9,6 +9,7 @@ defmodule Supavisor.SecretChecker do
   alias Supavisor.ClientAuthentication
   alias Supavisor.ClientAuthentication.ValidationSecrets
   alias Supavisor.Errors.AuthQueryError
+  alias Supavisor.Monitoring.Telem
 
   @interval :timer.seconds(15)
 
@@ -40,7 +41,10 @@ defmodule Supavisor.SecretChecker do
             {:error, %AuthQueryError{reason: :timeout}}
 
           :exit, reason ->
-            Logger.error("SecretChecker: get_secrets call exited: #{inspect(reason)}")
+            Logger.error("SecretChecker: get_secrets call exited: #{inspect(reason)}",
+              project: Supavisor.id(id, :tenant)
+            )
+
             {:error, :not_started}
         end
     end
@@ -161,9 +165,12 @@ defmodule Supavisor.SecretChecker do
   defp jitter, do: :rand.uniform(div(@interval, 10))
 
   defp erpc_call_node(id, mod, fun, args) do
+    start = System.monotonic_time()
+
     case Supavisor.get_global_sup(id) do
       nil ->
         {:error, :not_started}
+        |> tap(&telemetry_stop(fun, start, &1, :local))
 
       pid ->
         try do
@@ -173,6 +180,27 @@ defmodule Supavisor.SecretChecker do
             Logger.error("SecretChecker: erpc call exited: #{inspect(reason)}")
             {:error, :not_started}
         end
+        |> tap(&telemetry_stop(fun, start, &1, (node(pid) == node() && :local) || :remote))
     end
   end
+
+  defp telemetry_stop(:do_get_secrets, start, result, locality) do
+    Telem.secret_checker_get_secrets_stop(
+      System.monotonic_time() - start,
+      telemetry_status(result),
+      locality
+    )
+  end
+
+  defp telemetry_stop(:do_update_credentials, start, result, locality) do
+    Telem.secret_checker_update_credentials_stop(
+      System.monotonic_time() - start,
+      telemetry_status(result),
+      locality
+    )
+  end
+
+  defp telemetry_status({:error, _}), do: :error
+  defp telemetry_status({:ok, _}), do: :ok
+  defp telemetry_status(:ok), do: :ok
 end