diff --git a/Makefile b/Makefile index 45ee0b4a..389abf28 100644 --- a/Makefile +++ b/Makefile @@ -11,6 +11,7 @@ dev: FLY_ALLOC_ID=111e4567-e89b-12d3-a456-426614174000 \ SECRET_KEY_BASE="dev" \ CLUSTER_POSTGRES="true" \ + DB_POOL_SIZE="5" \ ERL_AFLAGS="-kernel shell_history enabled" \ iex --name node1@127.0.0.1 --cookie cookie -S mix run --no-halt diff --git a/VERSION b/VERSION index c5676407..13fec672 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.1.33 +1.1.35 diff --git a/config/runtime.exs b/config/runtime.exs index 3d7b83c6..5f7113c7 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -158,7 +158,7 @@ if config_env() != :test do config :supavisor, Supavisor.Repo, url: System.get_env("DATABASE_URL", "ecto://postgres:postgres@localhost:6432/postgres"), - pool_size: System.get_env("DB_POOL_SIZE", "5") |> String.to_integer(), + pool_size: System.get_env("DB_POOL_SIZE", "50") |> String.to_integer(), parameters: [ application_name: "supavisor_meta" ] diff --git a/lib/supavisor.ex b/lib/supavisor.ex index b926fd78..5fbbae10 100644 --- a/lib/supavisor.ex +++ b/lib/supavisor.ex @@ -141,12 +141,23 @@ defmodule Supavisor do |> :erpc.multicall(Supavisor, :dirty_terminate, [tenant], 60_000) end - @spec del_all_cache(String.t(), String.t()) :: map() + @spec del_all_cache(String.t(), String.t()) :: [map()] def del_all_cache(tenant, user) do - %{ - secrets: Cachex.del(Supavisor.Cache, {:secrets, tenant, user}), - metrics: Cachex.del(Supavisor.Cache, {:metrics, tenant}) - } + Logger.info("Deleting all cache for tenant #{tenant} and user #{user}") + {:ok, keys} = Cachex.keys(Supavisor.Cache) + + del = fn key, acc -> + result = Cachex.del(Supavisor.Cache, key) + [%{inspect(key) => inspect(result)} | acc] + end + + Enum.reduce(keys, [], fn + {:metrics, ^tenant} = key, acc -> del.(key, acc) + {:secrets, ^tenant, ^user} = key, acc -> del.(key, acc) + {:user_cache, _, ^user, ^tenant, _} = key, acc -> del.(key, acc) + {:tenant_cache, ^tenant, _} = key, acc -> del.(key, acc) + _, acc -> acc + end) end @spec get_local_pool(id) :: map | pid | nil diff --git a/lib/supavisor/application.ex b/lib/supavisor/application.ex index 788f0ce1..0a90c5e5 100644 --- a/lib/supavisor/application.ex +++ b/lib/supavisor/application.ex @@ -57,6 +57,7 @@ defmodule Supavisor.Application do topologies = Application.get_env(:libcluster, :topologies) || [] children = [ + Supavisor.ErlSysMon, PromEx, {Registry, keys: :unique, name: Supavisor.Registry.Tenants}, {Registry, keys: :unique, name: Supavisor.Registry.ManagerTables}, diff --git a/lib/supavisor/client_handler.ex b/lib/supavisor/client_handler.ex index 039a0384..8d373345 100644 --- a/lib/supavisor/client_handler.ex +++ b/lib/supavisor/client_handler.ex @@ -219,18 +219,22 @@ defmodule Supavisor.ClientHandler do true -> new_data = update_user_data(data, info, user, id, db_name, mode) - case auth_secrets(info, user) do + key = {:secrets, tenant_or_alias, user} + + case auth_secrets(info, user, key, :timer.hours(24)) do {:ok, auth_secrets} -> Logger.debug("ClientHandler: Authentication method: #{inspect(auth_secrets)}") - {:keep_state, new_data, {:next_event, :internal, {:handle, auth_secrets}}} + {:keep_state, new_data, {:next_event, :internal, {:handle, auth_secrets, info}}} {:error, reason} -> Logger.error( "ClientHandler: Authentication auth_secrets error: #{inspect(reason)}" ) - :ok = HH.send_error(sock, "XX000", "Authentication error") + :ok = + HH.send_error(sock, "XX000", "Authentication error, reason: #{inspect(reason)}") + Telem.client_join(:fail, id) {:stop, {:shutdown, :auth_secrets_error}} end @@ -249,7 +253,7 @@ defmodule Supavisor.ClientHandler do def handle_event( :internal, - {:handle, {method, secrets}}, + {:handle, {method, secrets}, info}, _, %{sock: sock} = data ) do @@ -268,6 +272,33 @@ defmodule Supavisor.ClientHandler do Server.exchange_message(:final, "e=#{reason}") end + key = {:secrets_check, data.tenant, data.user} + + if method != :password and reason == "Wrong password" and + Cachex.get(Supavisor.Cache, key) == {:ok, nil} do + case auth_secrets(info, data.user, key, 15_000) do + {:ok, {method2, secrets2}} = value -> + if method != method2 || Map.delete(secrets.(), :client_key) != secrets2.() do + Logger.warning("ClientHandler: Update secrets and terminate pool") + + Cachex.update( + Supavisor.Cache, + {:secrets, data.tenant, data.user}, + {:cached, value} + ) + + Supavisor.stop(data.id) + else + Logger.debug("ClientHandler: Cache the same #{inspect(key)}") + end + + other -> + Logger.error("ClientHandler: Auth secrets check error: #{inspect(other)}") + end + else + Logger.debug("ClientHandler: Cache hit for #{inspect(key)}") + end + HH.sock_send(sock, msg) Telem.client_join(:fail, data.id) {:stop, {:shutdown, :exchange_error}} @@ -469,11 +500,14 @@ defmodule Supavisor.ClientHandler do "ClientHandler: Manager #{inspect(data.manager)} went down #{inspect(reason)} state #{inspect(state)}" ) - case state do - :idle -> + case {state, reason} do + {_, :shutdown} -> + {:stop, {:shutdown, :manager_shutdown}} + + {:idle, _} -> {:keep_state_and_data, {:next_event, :internal, :subscribe}} - :busy -> + {:busy, _} -> {:stop, {:shutdown, :manager_down}} end end @@ -730,26 +764,25 @@ defmodule Supavisor.ClientHandler do } end - @spec auth_secrets(map, String.t()) :: {:ok, S.secrets()} | {:error, term()} + @spec auth_secrets(map, String.t(), term(), non_neg_integer()) :: + {:ok, S.secrets()} | {:error, term()} ## password secrets - def auth_secrets(%{user: user, tenant: %{require_user: true}}, _) do + def auth_secrets(%{user: user, tenant: %{require_user: true}}, _, _, _) do secrets = %{db_user: user.db_user, password: user.db_password, alias: user.db_user_alias} {:ok, {:password, fn -> secrets end}} end ## auth_query secrets - def auth_secrets(%{tenant: tenant} = info, db_user) do - cache_key = {:secrets, tenant.external_id, db_user} - + def auth_secrets(info, db_user, key, ttl) do fetch = fn _key -> case get_secrets(info, db_user) do - {:ok, _} = resp -> {:commit, {:cached, resp}, ttl: 600_000} + {:ok, _} = resp -> {:commit, {:cached, resp}, ttl: ttl} {:error, _} = resp -> {:ignore, resp} end end - case Cachex.fetch(Supavisor.Cache, cache_key, fetch) do + case Cachex.fetch(Supavisor.Cache, key, fetch) do {:ok, {:cached, value}} -> value {:commit, {:cached, value}, _opts} -> value {:ignore, resp} -> resp diff --git a/lib/supavisor/db_handler.ex b/lib/supavisor/db_handler.ex index b0072982..8cc776c9 100644 --- a/lib/supavisor/db_handler.ex +++ b/lib/supavisor/db_handler.ex @@ -291,7 +291,7 @@ defmodule Supavisor.DbHandler do # the process received message from db without linked caller def handle_event(:info, {proto, _, bin}, _, %{caller: nil}) when proto in @proto do - Logger.warning("DbHandler: Got db response #{inspect(bin)} when caller was nil") + Logger.debug("DbHandler: Got db response #{inspect(bin)} when caller was nil") :keep_state_and_data end diff --git a/lib/supavisor/erl_sys_mon.ex b/lib/supavisor/erl_sys_mon.ex new file mode 100644 index 00000000..d6fff155 --- /dev/null +++ b/lib/supavisor/erl_sys_mon.ex @@ -0,0 +1,30 @@ +defmodule Supavisor.ErlSysMon do + @moduledoc """ + Logs Erlang System Monitor events. + """ + + use GenServer + + require Logger + + def start_link(args) do + GenServer.start_link(__MODULE__, args, name: __MODULE__) + end + + def init(_args) do + :erlang.system_monitor(self(), [ + :busy_dist_port, + :busy_port, + {:long_gc, 250}, + {:long_schedule, 100} + ]) + + {:ok, []} + end + + def handle_info(msg, state) do + Logger.warning("#{__MODULE__} message: " <> inspect(msg)) + + {:noreply, state} + end +end diff --git a/lib/supavisor/tenants.ex b/lib/supavisor/tenants.ex index 54272d89..26095c67 100644 --- a/lib/supavisor/tenants.ex +++ b/lib/supavisor/tenants.ex @@ -55,7 +55,7 @@ defmodule Supavisor.Tenants do cache_key = {:tenant_cache, external_id, sni_hostname} case Cachex.fetch(Supavisor.Cache, cache_key, fn _key -> - {:commit, {:cached, get_tenant(external_id, sni_hostname)}, ttl: 5_000} + {:commit, {:cached, get_tenant(external_id, sni_hostname)}, ttl: :timer.hours(24)} end) do {_, {:cached, value}} -> value {_, {:cached, value}, _} -> value @@ -79,7 +79,8 @@ defmodule Supavisor.Tenants do cache_key = {:user_cache, type, user, external_id, sni_hostname} case Cachex.fetch(Supavisor.Cache, cache_key, fn _key -> - {:commit, {:cached, get_user(type, user, external_id, sni_hostname)}, ttl: 5_000} + {:commit, {:cached, get_user(type, user, external_id, sni_hostname)}, + ttl: :timer.hours(24)} end) do {_, {:cached, value}} -> value {_, {:cached, value}, _} -> value