diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index 0f4dd7b1..dd7b1712 100644 --- a/grpc/lib/grpc/client/connection.ex +++ b/grpc/lib/grpc/client/connection.ex @@ -97,6 +97,12 @@ defmodule GRPC.Client.Connection do shut down, after its resources were released. * Measurements: none * Metadata: `:name`, `:target`, `:pid`, `:reason` + * `[:grpc, :client, :connection, :channel_pruned]` – a channel's adapter + connection process died and the channel was removed from the load + balancer, so it is no longer picked. + * Measurements: `:remaining` – channels still connected afterwards + * Metadata: `:name`, `:target`, `:pid`, `:reason`, `:address` – the + address key of the pruned channel * `[:grpc, :client, :connection, :await_ready, :start]` – a caller started waiting in `await_ready/2`. * Measurements: `:system_time` @@ -163,6 +169,7 @@ defmodule GRPC.Client.Connection do @connected_event [:grpc, :client, :connection, :connected] @connect_error_event [:grpc, :client, :connection, :connect_error] @disconnected_event [:grpc, :client, :connection, :disconnected] + @channel_pruned_event [:grpc, :client, :connection, :channel_pruned] @await_ready_start_event [:grpc, :client, :connection, :await_ready, :start] @await_ready_stop_event [:grpc, :client, :connection, :await_ready, :stop] @@ -573,20 +580,12 @@ defmodule GRPC.Client.Connection do {:noreply, state} else - Logger.warning( - "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}" - ) - - {:noreply, state} + {:noreply, handle_adapter_exit(pid, reason, state)} end end def handle_info({:EXIT, pid, reason}, state) do - Logger.warning( - "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}" - ) - - {:noreply, state} + {:noreply, handle_adapter_exit(pid, reason, state)} end def handle_info({:DOWN, mon, :process, pid, reason}, state) do @@ -937,7 +936,7 @@ defmodule GRPC.Client.Connection do end if connected == [] do - Logger.warning("No healthy channels available after re-resolution") + Logger.warning("No healthy channels available") end %{state | real_channels: real_channels, lb_state: new_lb_state} @@ -958,6 +957,65 @@ defmodule GRPC.Client.Connection do defp channel_alive?({:connected, _}), do: true defp channel_alive?(_), do: false + # An adapter connection process died. It is linked to us, so we are the only + # one told: the channel it backs is still `{:connected, ch}` in + # `real_channels` and still in the load balancer's list, so `pick_channel/2` + # would keep handing out a channel whose `conn_pid` is dead until the next + # re-resolution reconciles it. Literal-address targets (`ipv4:`, `ipv6:`, + # `unix:`) have no background re-resolution at all, so for those no tick is + # ever coming and the dead channel would be served forever. + defp handle_adapter_exit(pid, reason, state) do + case channel_key_for_conn_pid(state.real_channels, pid) do + nil -> + Logger.warning( + "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}" + ) + + state + + key -> + Logger.warning( + "#{inspect(__MODULE__)} adapter connection #{inspect(pid)} exited: " <> + "#{inspect(reason)}, pruning its channel from the load balancer" + ) + + new_state = + state.real_channels + |> Map.put(key, {:failed, reason}) + |> rebalance_after_reconcile(state) + + :telemetry.execute( + @channel_pruned_event, + %{remaining: length(connected_channels(new_state.real_channels))}, + state + |> lifecycle_metadata() + |> Map.merge(%{reason: reason, address: key}) + ) + + maybe_reestablish(new_state) + end + end + + defp channel_key_for_conn_pid(real_channels, pid) do + Enum.find_value(real_channels, fn + {key, {:connected, %{adapter_payload: %{conn_pid: ^pid}}}} -> key + _ -> nil + end) + end + + # With channels left we simply serve them and let the next reconcile retry + # the failed address. With none left there is nothing to pick, so restart the + # establish loop from scratch -- `:retry_establish` is a no-op while + # `established?` is true, so that has to be cleared for the retry to run. + defp maybe_reestablish(state) do + if connected_channels(state.real_channels) == [] do + Process.send_after(self(), :retry_establish, backoff_delay(0)) + %{state | established?: false, retry_attempt: 0} + else + state + end + end + defp via(ref) do {:via, Registry, {GRPC.Client.Registry, {__MODULE__, ref}}} end diff --git a/grpc/test/grpc/client/connection_supervised_test.exs b/grpc/test/grpc/client/connection_supervised_test.exs index 3eceb0b5..996f26a2 100644 --- a/grpc/test/grpc/client/connection_supervised_test.exs +++ b/grpc/test/grpc/client/connection_supervised_test.exs @@ -145,6 +145,96 @@ defmodule GRPC.Client.ConnectionSupervisedTest do end end + describe "adapter connection process exits" do + @tag capture_log: true + test "stops picking a channel whose adapter process died" do + name = unique_name("dead_channel") + attach_telemetry([:grpc, :client, :connection, :channel_pruned]) + + start_supervised!( + {Connection, + name: name, + target: "ipv4:127.0.0.1:50051,127.0.0.1:50052", + adapter: GRPC.Test.ProcessClientAdapter} + ) + + assert :ok = Connection.await_ready(name, 2_000) + + assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: dead_pid}}} = + Connection.pick_channel(%GRPC.Channel{ref: name}) + + Process.exit(dead_pid, :kill) + + # The exit signal comes from the dying process, so it is not ordered + # against anything this process sends. Sync on the prune itself. + assert_receive {:telemetry, [:grpc, :client, :connection, :channel_pruned], %{remaining: 1}, + %{name: ^name, reason: :killed}}, + 1_000 + + refute Process.alive?(dead_pid) + + assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: live_pid}}} = + Connection.pick_channel(%GRPC.Channel{ref: name}) + + assert live_pid != dead_pid + assert Process.alive?(live_pid) + end + + @tag capture_log: true + test "re-establishes when the last remaining channel's adapter process dies" do + name = unique_name("last_channel") + attach_telemetry([:grpc, :client, :connection, :connected]) + + start_supervised!( + {Connection, + name: name, target: "ipv4:127.0.0.1:50051", adapter: GRPC.Test.ProcessClientAdapter} + ) + + assert :ok = Connection.await_ready(name, 2_000) + assert_receive {:telemetry, [:grpc, :client, :connection, :connected], _, %{name: ^name}} + + assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: dead_pid}}} = + Connection.pick_channel(%GRPC.Channel{ref: name}) + + Process.exit(dead_pid, :kill) + + # The only channel is gone, so the establish loop has to run again. A + # second :connected event can only come from that. + assert_receive {:telemetry, [:grpc, :client, :connection, :connected], _, %{name: ^name}}, + 2_000 + + assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: live_pid}}} = + Connection.pick_channel(%GRPC.Channel{ref: name}) + + assert live_pid != dead_pid + assert Process.alive?(live_pid) + end + + @tag capture_log: true + test "an unrelated linked exit leaves the channels alone" do + name = unique_name("unrelated_exit") + + start_supervised!( + {Connection, + name: name, target: "ipv4:127.0.0.1:50051", adapter: GRPC.Test.ProcessClientAdapter} + ) + + assert :ok = Connection.await_ready(name, 2_000) + + assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid}}} = + Connection.pick_channel(%GRPC.Channel{ref: name}) + + conn = whereis_connection(name) + send(conn, {:EXIT, spawn(fn -> :ok end), :some_crash}) + :sys.get_state(conn) + + assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: ^pid}}} = + Connection.pick_channel(%GRPC.Channel{ref: name}) + + assert Process.alive?(pid) + end + end + describe "await_ready/2 waiter lifecycle" do setup do %{ diff --git a/grpc/test/support/test_adapter.exs b/grpc/test/support/test_adapter.exs index 03faafbe..b5b75c91 100644 --- a/grpc/test/support/test_adapter.exs +++ b/grpc/test/support/test_adapter.exs @@ -11,6 +11,43 @@ defmodule GRPC.Test.ClientAdapter do def cancel(stream), do: stream end +defmodule GRPC.Test.ProcessClientAdapter do + @moduledoc """ + A test adapter that backs each channel with a real linked process recorded + as `conn_pid`, the way the Gun and Mint adapters do. + + `GRPC.Test.ClientAdapter` returns the channel untouched, so its channels have + no `conn_pid` and there is no process to kill. This adapter exists so tests + can kill the process behind a channel and assert on what the connection does + about it. + """ + @behaviour GRPC.Client.Adapter + + def connect(channel, _opts) do + {:ok, %{channel | adapter_payload: %{conn_pid: spawn_link(&idle/0)}}} + end + + def disconnect(%{adapter_payload: %{conn_pid: pid}} = channel) when is_pid(pid) do + send(pid, :stop) + {:ok, %{channel | adapter_payload: %{conn_pid: nil}}} + end + + def disconnect(channel), do: {:ok, channel} + + defp idle do + receive do + :stop -> :ok + end + end + + def send_request(stream, _message, _opts), do: stream + def receive_data(_stream, _opts), do: {:ok, nil} + def send_data(stream, _message, _opts), do: stream + def send_headers(stream, _opts), do: stream + def end_stream(stream), do: stream + def cancel(stream), do: stream +end + defmodule GRPC.Test.FailingClientAdapter do @moduledoc """ A test adapter that refuses to connect to selected hosts. All other hosts