diff --git a/fluxer_gateway/src/gateway/fluxer_gateway_app.erl b/fluxer_gateway/src/gateway/fluxer_gateway_app.erl index cc2ae6979..7a26b1cfd 100644 --- a/fluxer_gateway/src/gateway/fluxer_gateway_app.erl +++ b/fluxer_gateway/src/gateway/fluxer_gateway_app.erl @@ -45,6 +45,7 @@ start_cowboy() -> {<<"/_health">>, health_handler, liveness}, {<<"/_health/ready">>, health_handler, readiness}, {<<"/_health/drain">>, health_handler, drain}, + {<<"/_health/undrain">>, health_handler, undrain}, {<<"/_metrics">>, metrics_handler, []}, {<<"/">>, gateway_handler, []} ]} diff --git a/fluxer_gateway/src/gateway/gateway_cluster_handoff.erl b/fluxer_gateway/src/gateway/gateway_cluster_handoff.erl index e6447f3ee..7e88ab3f1 100644 --- a/fluxer_gateway/src/gateway/gateway_cluster_handoff.erl +++ b/fluxer_gateway/src/gateway/gateway_cluster_handoff.erl @@ -17,6 +17,7 @@ -define(DEBOUNCE_MS, 2000). -define(RECONCILE_MS, 60000). -define(HANDOFF_WORKER_TIMEOUT_MS, 30000). +-define(UNDRAIN_CALL_TIMEOUT_MS, 5000). -type timer_state() :: undefined | {reference(), reference()}. -type reconcile_timer() :: undefined | reference(). @@ -50,8 +51,17 @@ drain_async() -> ok = drain_notify_role(), drain_dispatch(). --spec undrain() -> ok. +-spec undrain() -> ok | {error, handoff_in_flight | unavailable}. undrain() -> + case whereis(?MODULE) of + undefined -> + clear_draining(); + _Pid -> + shard_utils:safe_gen_call(?MODULE, undrain, ?UNDRAIN_CALL_TIMEOUT_MS) + end. + +-spec clear_draining() -> ok. +clear_draining() -> persistent_term:erase({fluxer_gateway, draining}), logger:info("Gateway un-cordoned: draining flag cleared"), ok. @@ -89,6 +99,10 @@ init([]) -> {reply, term(), state()}. handle_call(diagnostic_info, _From, State) -> {reply, info(State), State}; +handle_call(undrain, _From, #{handoff := undefined} = State) -> + {reply, clear_draining(), State}; +handle_call(undrain, _From, State) -> + {reply, {error, handoff_in_flight}, State}; handle_call(_Request, _From, State) -> {reply, ok, State}. @@ -615,6 +629,18 @@ undrain_clears_draining_flag_test() -> ?assertEqual(false, persistent_term:get({fluxer_gateway, draining}, false)), restore_persistent_term({fluxer_gateway, draining}, Previous). +undrain_is_refused_while_handoff_in_flight_test() -> + Previous = persistent_term:get({fluxer_gateway, draining}, undefined), + persistent_term:put({fluxer_gateway, draining}, true), + Busy = (idle_state([node()]))#{handoff := {self(), make_ref(), [node()], #{}}}, + {reply, Refused, Busy} = handle_call(undrain, {self(), make_ref()}, Busy), + ?assertEqual({error, handoff_in_flight}, Refused), + ?assertEqual(true, persistent_term:get({fluxer_gateway, draining}, false)), + {reply, Cleared, _Idle} = handle_call(undrain, {self(), make_ref()}, idle_state([node()])), + ?assertEqual(ok, Cleared), + ?assertEqual(false, persistent_term:get({fluxer_gateway, draining}, false)), + restore_persistent_term({fluxer_gateway, draining}, Previous). + handoff_wait_loop() -> receive stop -> ok diff --git a/fluxer_gateway/src/gateway/health_handler.erl b/fluxer_gateway/src/gateway/health_handler.erl index f0930bc67..a6cae9d4c 100644 --- a/fluxer_gateway/src/gateway/health_handler.erl +++ b/fluxer_gateway/src/gateway/health_handler.erl @@ -5,7 +5,7 @@ -export([init/2]). --type mode() :: liveness | readiness | drain. +-type mode() :: liveness | readiness | drain | undrain. -spec init(cowboy_req:req(), term()) -> {ok, cowboy_req:req(), term()}. init(Req0, Mode0) -> @@ -21,16 +21,19 @@ init(Req0, Mode0) -> -spec normalize_mode(term()) -> mode(). normalize_mode(drain) -> drain; +normalize_mode(undrain) -> undrain; normalize_mode(readiness) -> readiness; normalize_mode(_) -> liveness. --spec response_for_mode(mode(), cowboy_req:req()) -> {200 | 403 | 503, binary()}. +-spec response_for_mode(mode(), cowboy_req:req()) -> {200 | 403 | 409 | 503, binary()}. response_for_mode(liveness, _Req) -> {200, <<"OK">>}; response_for_mode(readiness, Req) -> readiness_response(Req); response_for_mode(drain, Req) -> - drain_response(Req). + drain_response(Req); +response_for_mode(undrain, Req) -> + undrain_response(Req). -spec readiness_response(cowboy_req:req()) -> {200 | 403 | 503, binary()}. readiness_response(Req) -> @@ -57,10 +60,31 @@ drain_response(Req) -> {200, <<"DRAINING">>} end. +-spec undrain_response(cowboy_req:req()) -> {200 | 403 | 409, binary()}. +undrain_response(Req) -> + case is_loopback_request(Req) of + false -> + {403, <<"FORBIDDEN">>}; + true -> + undrain_status(deactivate_drain()) + end. + +-spec undrain_status(term()) -> {200 | 409, binary()}. +undrain_status(ok) -> + {200, <<"READY">>}; +undrain_status({error, handoff_in_flight}) -> + {409, <<"HANDOFF_IN_FLIGHT">>}; +undrain_status(_Error) -> + {409, <<"UNAVAILABLE">>}. + -spec activate_drain() -> ok. activate_drain() -> gateway_cluster_handoff:drain_async(). +-spec deactivate_drain() -> ok | {error, term()}. +deactivate_drain() -> + gateway_cluster_handoff:undrain(). + -spec is_loopback_request(cowboy_req:req()) -> boolean(). is_loopback_request(Req) -> case cowboy_req:peer(Req) of @@ -79,16 +103,31 @@ normalize_mode_test() -> ?assertEqual(liveness, normalize_mode(undefined)), ?assertEqual(liveness, normalize_mode([])), ?assertEqual(readiness, normalize_mode(readiness)), - ?assertEqual(drain, normalize_mode(drain)). + ?assertEqual(drain, normalize_mode(drain)), + ?assertEqual(undrain, normalize_mode(undrain)). readiness_status_test() -> ?assertEqual({200, <<"OK">>}, readiness_status(true)), ?assertEqual({503, <<"DRAINING">>}, readiness_status(false)). +undrain_status_test() -> + ?assertEqual({200, <<"READY">>}, undrain_status(ok)), + ?assertEqual({409, <<"HANDOFF_IN_FLIGHT">>}, undrain_status({error, handoff_in_flight})), + ?assertEqual({409, <<"UNAVAILABLE">>}, undrain_status({error, unavailable})). + activate_drain_sets_draining_flag_test() -> persistent_term:erase({fluxer_gateway, draining}), ?assertEqual(ok, activate_drain()), ?assert(gateway_node_router:is_draining()), persistent_term:erase({fluxer_gateway, draining}). +deactivate_drain_clears_draining_flag_test() -> + persistent_term:erase({fluxer_gateway, draining}), + ?assertEqual(ok, activate_drain()), + ?assert(gateway_node_router:is_draining()), + ?assertEqual(ok, deactivate_drain()), + ?assertNot(gateway_node_router:is_draining()), + ?assertEqual({200, <<"OK">>}, readiness_status(gateway_node_router:is_ready())), + persistent_term:erase({fluxer_gateway, draining}). + -endif. diff --git a/fluxer_gateway/src/guild/guild_ets_owner.erl b/fluxer_gateway/src/guild/guild_ets_owner.erl index 3d4febdea..4c0913a21 100644 --- a/fluxer_gateway/src/guild/guild_ets_owner.erl +++ b/fluxer_gateway/src/guild/guild_ets_owner.erl @@ -8,8 +8,9 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(DEFAULT_CALL_TIMEOUT_MS, 5000). +-define(ORPHAN_SWEEP_INTERVAL_MS, 60000). --type state() :: #{}. +-type state() :: #{sweep_timer => reference()}. -spec start_link() -> {ok, pid()} | {error, term()}. start_link() -> @@ -32,7 +33,7 @@ ensure_table(TableName, Options) -> init([]) -> erlang:process_flag(fullsweep_after, 0), ok = ensure_core_tables(), - {ok, #{}}. + {ok, #{sweep_timer => schedule_orphan_sweep()}}. -spec handle_call(term(), gen_server:from(), state()) -> {reply, ok, state()}. handle_call({ensure_table, TableName, Options}, _From, State) when @@ -47,11 +48,26 @@ handle_cast(_Msg, State) -> {noreply, State}. -spec handle_info(term(), state()) -> {noreply, state()}. +handle_info(sweep_orphan_tables, State) -> + _ = sweep_orphan_tables(), + {noreply, State#{sweep_timer => schedule_orphan_sweep()}}; handle_info(_Info, State) -> {noreply, State}. +-spec schedule_orphan_sweep() -> reference(). +schedule_orphan_sweep() -> + erlang:send_after(?ORPHAN_SWEEP_INTERVAL_MS, self(), sweep_orphan_tables). + -spec terminate(term(), state()) -> ok. -terminate(_Reason, _State) -> +terminate(_Reason, State) -> + cancel_orphan_sweep(State), + ok. + +-spec cancel_orphan_sweep(state()) -> ok. +cancel_orphan_sweep(#{sweep_timer := Ref}) when is_reference(Ref) -> + _ = erlang:cancel_timer(Ref), + ok; +cancel_orphan_sweep(_State) -> ok. -spec code_change(term(), state(), term()) -> {ok, state()}. @@ -229,6 +245,20 @@ sweep_orphan_tables_detects_dead_owner_test() -> ?assert(Count >= 0), ets:delete(Tab). +orphan_sweep_is_scheduled_and_rearmed_test() -> + {ok, Pid} = start_link(), + try + #{sweep_timer := Ref} = sys:get_state(Pid), + ?assert(is_reference(Ref)), + Pid ! sweep_orphan_tables, + #{sweep_timer := NextRef} = sys:get_state(Pid), + ?assert(is_reference(NextRef)), + ?assertNotEqual(Ref, NextRef), + ?assertEqual(Pid, ets:info(guild_voice_registry, owner)) + after + gen_server:stop(?MODULE) + end. + sweep_orphan_tables_returns_zero_when_clean_test() -> {ok, Count} = sweep_orphan_tables(), ?assert(is_integer(Count)),