Merge pull request #6 from emqx/fix/stale-grpc-channel-recovery-emqx

fix: recover stale grpc channels

Author: hjianbo

src/greptimedb_worker.erl

@@ -71,7 +71,7 @@ init(Args) ->
         lists:map(fun({Scheme, Host, Port}) -> {Scheme, Host, Port, ssl_options(Scheme, SslOptions)}
                   end, Endpoints),
     Channel = iolist_to_binary([PoolName, ":", integer_to_binary(WorkerId)]),
-    {ok, _} = grpcbox_channel_sup:start_child(Channel, Channels, Options),
+    {ok, _} = start_channel(Channel, Channels, Options),
     logger:debug("[GreptimeDB] genserver has started (~s)~n", [Channel]),
     {ok, #state{channel = Channel, hints = Hints, requests = #{ pending => queue:new(), pending_count => 0}}}.
 
@@ -134,6 +134,28 @@ terminate(Reason, #state{channel = Channel} = State) ->
 %%%===================================================================
 %%% Helper functions
 %%%===================================================================
+start_channel(Channel, Channels, Options) ->
+    case grpcbox_channel_sup:start_child(Channel, Channels, Options) of
+        {error, {already_started, StaleChannel}} ->
+            %% ecpool may force-kill a worker without invoking terminate/2, leaving its
+            %% grpcbox channel alive under the global supervisor. Remove that orphan
+            %% before reusing the deterministic channel name.
+            logger:warning(
+              "[GreptimeDB] removing stale grpc channel ~s (~p)",
+              [Channel, StaleChannel]),
+            ok = stop_stale_channel(Channel),
+            grpcbox_channel_sup:start_child(Channel, Channels, Options);
+        Result ->
+            Result
+    end.
+
+stop_stale_channel(Channel) ->
+    try grpcbox_channel:stop(Channel)
+    catch
+        exit:noproc -> ok;
+        exit:{noproc, _} -> ok
+    end.
+
 ssl_options(https, []) ->
     %% https://www.erlang.org/doc/man/ssl#type-client_option
     [

test/greptimedb_SUITE.erl

@@ -13,7 +13,8 @@
 -define(WRONG_PASSWORD, <<"wrong_pwd">>).
 
 all() ->
-    [t_write,
+    [t_recover_stale_channel,
+     t_write,
      t_write_stream,
      t_write_failure,
      t_write_batch,
@@ -41,6 +42,29 @@ all() ->
      t_write_custom_ts_column,
      t_write_decimal128].
 
+t_recover_stale_channel(_) ->
+    Pool = greptimedb_stale_channel_pool,
+    Channel = <<"greptimedb_stale_channel_pool:1">>,
+    Endpoint = {http, greptime_host(), 5001},
+    GrpcOptions = #{connect_timeout => 5_000},
+    Options =
+        [{endpoints, [Endpoint]},
+         {pool, Pool},
+         {pool_size, 1},
+         {grpc_opts, GrpcOptions}],
+    try
+        {Scheme, Host, Port} = Endpoint,
+        {ok, StaleChannel} =
+            grpcbox_channel_sup:start_child(
+              Channel, [{Scheme, Host, Port, []}], GrpcOptions),
+        {ok, Client} = greptimedb:start_client(Options),
+        ?assertNot(is_process_alive(StaleChannel)),
+        ok = greptimedb:stop_client(Client)
+    after
+        _ = catch ecpool:stop_sup_pool(Pool),
+        _ = catch grpcbox_channel:stop(Channel)
+    end.
+
 %%[t_bench_perf].
 %%[t_insert_requests, t_bench_perf].