fuse

%%% @doc Runs the fuse server in the system
%%% @private
-module(fuse_server).
-behaviour(gen_server).

-ifdef(PULSE).
-include_lib("pulse_otp/include/pulse_otp.hrl").
-endif.


%% Lifetime API
-export([start_link/0]).

%% Operational API
-export([
    ask/2,
    install/2,
    melt/1,
    remove/1,
    reset/1,
    run/3]).

%% Administrative API
-export([
    circuit/2
]).

%% Callbacks
-export([code_change/3, handle_call/3, handle_cast/2, handle_info/2, init/1, terminate/2]).

%% Private
-export([sync/0, q_melts/0]).

-define(TAB, fuse_state).

-record(state, { fuses = [] }).
-record(fuse, {
    name :: atom(),
    intensity :: integer(),
    period :: integer(),
    heal_time :: integer(),
    melt_history = [],
    ty = ok :: 'ok' | {gradual, float()},
    timer_ref = none,
    enabled = true
}).

%% -- API -----------------------------------------------------

%% ------
%% @doc Start up the manager server for the fuse system
%% This is assumed to be called by (@see fuse_sup). The `Timing' parameter controls how the system manages timing.
%% @end
start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

%% ------
%% @doc install/2 installs a new fuse into the running system
%% Install a new fuse under `Name' with options given by `Opts'.
%% We assume `Opts' are already in the right verified and validated format.
%% @end
install(Name, Opts) ->
    %% Assume options are already verified
    Fuse = init_state(Name, Opts),
    gen_server:call(?MODULE, {install, Fuse}).

%% @doc ask/2 asks about the current given fuse state in a given context setting
%% The documentation is (@see fuse:ask/1)
%% @end
-spec ask(atom(), fuse:fuse_context()) -> ok | blown | {error, not_found}.
ask(Name, sync) -> gen_server:call(?MODULE, {ask, Name});
ask(Name, async_dirty) -> ask_(Name).

%% ask_/1 is the real ask function.
ask_(Name) ->
    StatsPlugin = application:get_env(fuse, stats_plugin, fuse_stats_ets),
    try ets:lookup_element(?TAB, Name, 2) of
        ok ->
          _ = StatsPlugin:increment(Name, ok),
          ok;
        blown ->
          _ = StatsPlugin:increment(Name, blown),
          blown;
        {gradual, Rate} ->
            case fuse_rand:uniform() of
                K when K < Rate ->
                    blown;
                _ ->
                    ok
            end
    catch
        error:badarg ->
            {error, not_found}
    end.

%% @doc reset/1 resets the fuse
%% The documentation is (@see fuse:reset/1)
%% @end
-spec reset(atom()) -> ok | {error, not_found}.
reset(Name) ->
    gen_server:call(?MODULE, {reset, Name}).

%% @doc circuit/2 is used to manually override the fuse state
%% @end
-spec circuit(atom(), enable | disable) -> ok.
circuit(Name, Switch) ->
    gen_server:call(?MODULE, {circuit, Name, Switch}).

%% @doc melt/2 melts the fuse at a given point in time
%% For documentation, (@see fuse:melt/2)
%% @end
-spec melt(Name) -> ok
    when Name :: atom().
melt(Name) ->
    gen_server:call(?MODULE, {melt, Name}).

%% @doc remove/1 removes the fuse
%% The documentation is (@see fuse:remove/1)
%% @end
-spec remove(atom()) -> ok | {error, not_found}.
remove(Name) ->
    gen_server:call(?MODULE, {remove, Name}).

%% sync/0 syncs the server. For internal use only in tests
%% @private
sync() ->
    gen_server:call(?MODULE, sync).

q_melts() ->
    gen_server:call(?MODULE, q_melts).

%% run/3 runs a thunk under a given fuse in a given context
%% @doc Documentation is (@see fuse:run/3)
%% @end
%% @private
-spec run(Name, fun(() -> {ok, Result} | {melt, Result}), fuse:fuse_context()) ->
    {ok, Result} | blown | {error, not_found}
  when
    Name :: atom(),
    Result :: any().

run(Name, Func, Context) ->
    case ask(Name, Context) of
        blown -> blown;
        ok ->
          case Func() of
              {ok, Result} -> {ok, Result};
              {melt, Result} ->
                  melt(Name),
                  {ok, Result}
          end;
        {error, Reason} ->
          {error, Reason}
    end.

%% -- CALLBACKS --------------------------------------------

%% @private
init([]) ->
    _ = ets:new(?TAB, [named_table, protected, set, {read_concurrency, true}, {keypos, 1}]),
    {ok, #state{ }}.

%% @private
handle_call({install, #fuse { name = Name } = F}, _From, #state { fuses = Fs } = State) ->
        case lists:keytake(Name, #fuse.name, Fs) of
            false ->
                install_metrics(F),
                fix(F),
                {reply, ok, State#state { fuses = lists:keystore(Name, #fuse.name, Fs, F)}};
            {value, OldFuse, _Otherfuses} ->
                EF = F#fuse { enabled = OldFuse#fuse.enabled },
                fix(EF),
                _ = reset_timer(OldFuse), %% For effect only
                %% Transplant the enabled-state from the old fuse
                {reply, ok, State#state {
                    fuses = lists:keystore(Name, #fuse.name, Fs, EF) }}
        end;
handle_call({circuit, Name, Switch}, _From, State) ->
    {Reply, State2} = handle_circuit(Name, Switch, State),
    {reply, Reply, State2};
handle_call({reset, Name}, _From, State) ->
    {Reply, State2} = handle_reset(Name, State, reset),
    {reply, Reply, State2};
handle_call({remove, Name}, _From, State) ->
    {Reply, State2} = handle_remove(Name, State),
    {reply, Reply, State2};
handle_call({melt, Name}, _From, State) ->
    Now = fuse_time:monotonic_time(),
    {Res, State2} = with_fuse(Name, State, fun(F) -> add_restart(Now, F) end),
    case Res of
      ok ->
            StatsPlugin = application:get_env(fuse, stats_plugin, fuse_stats_ets),
            _ = StatsPlugin:increment(Name, melt),
        {reply, ok, State2};
      not_found -> {reply, ok, State2}
    end;
handle_call({ask, Name}, _F, State) ->
    {reply, ask_(Name), State};
handle_call(sync, _F, State) ->
    {reply, ok, State};
handle_call(q_melts, _From, #state { fuses = Fs } = State) ->
        {reply, [{N, Ms} || #fuse { name = N, melt_history = Ms } <- Fs], State};
handle_call(q_disabled, _From, #state { fuses = Fs } = State) ->
    {reply, [Name || #fuse { name = Name, enabled = false } <- Fs], State};
handle_call(_M, _F, State) ->
    {reply, {error, unknown}, State}.

%% @private
handle_cast(_M, State) ->
    {noreply, State}.

%% @private
handle_info({reset, Name}, State) ->
    {_Reply, State2} = handle_reset(Name, State, timeout),
    {noreply, State2};
handle_info(_M, State) ->
    {noreply, State}.

%% @private
terminate(_Reason, _State) ->
    ok.

%% @private
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%% Internal functions
%%% ------

handle_reset(Name, State, ResetType) ->
    Reset = fun(#fuse { timer_ref = TRef } = F) ->
            case ResetType of
                reset ->
                    fix(F),
                    NewF = reset_timer(F),
                    {ok, NewF#fuse { melt_history = [] }};
                timeout when TRef /= none ->
                    fix(F),
                    NewF = reset_timer(F),
                    {ok, NewF#fuse { melt_history = [], timer_ref = none }};
                timeout when TRef == none ->
                    %% No timer installed on this fuse, it is a async fluke
                    {ok, F}
            end
    end,
    {Res, State2} = with_fuse(Name, State, Reset),
    case Res of
      ok -> {ok, State2};
      not_found -> {{error, not_found}, State2}
    end.

handle_circuit(Name, Switch, State) ->
    Fn = fun(#fuse { enabled = En } = F) ->
        case Switch of
            enable when En -> {ok, F};
            enable ->
                F1 = F#fuse { enabled = true },
                fix(F1),
                {ok, F1#fuse { melt_history = [] }};
            disable when not En -> {ok, F};
            disable ->
                blow(F),
                NewF = reset_timer(F),
                {ok, NewF#fuse { enabled = false }}
        end
    end,
    {Res, State2} = with_fuse(Name, State, Fn),
    case Res of
        ok -> {ok, State2};
        not_found -> {{error, not_found}, State2}
    end.

handle_remove(Name, #state { fuses = Fs } = State) ->
    case lists:keytake(Name, #fuse.name, Fs) of
        false -> {{error, not_found}, State};
        {value, F, OtherFs} ->
            delete(F),
            {ok, State#state { fuses = OtherFs }}
    end.

init_state(Name, {{fault_injection, Rate, MR, MT}, {reset, Reset}}) ->
    init_state(Name, {gradual, Rate}, MR, MT, {reset, Reset});
init_state(Name, {{standard, MR, MT}, {reset, Reset}}) ->
    init_state(Name, ok, MR, MT, {reset, Reset}).

init_state(Name, Ty, MR, MT, {reset, Reset}) ->
    NativePeriod = fuse_time:convert_time_unit(MT, milli_seconds, native),
    #fuse {
      name = Name,
      intensity = MR,
      period = NativePeriod,
      heal_time = Reset,
      ty = Ty
    }.

with_fuse(Name, #state { fuses = Fs} = State, Fun) ->
    case lists:keytake(Name, #fuse.name, Fs) of
        false -> {not_found, State};
        {value, F, OtherFs} ->
            {R, FF} = Fun(F),
            {R, State#state { fuses = [FF | OtherFs] }}
    end.

add_restart(Now, #fuse { intensity = I, period = Period, melt_history = R } = Fuse) ->
    R1 = add_restart_([Now | R], Now, Period),
    NewF = Fuse#fuse { melt_history = R1 },
    case length(R1) of
        CurI when CurI =< I ->
            {ok, NewF};
        _ ->
            blow(NewF),
            {ok, install_timer(NewF)}
    end.

add_restart_([R|Restarts], Now, Period) ->
    case in_period(R, Now, Period) of
        true -> [R|add_restart_(Restarts, Now, Period)];
        false -> []
    end;
add_restart_([], _, _) -> [].

in_period(Time, Now, Period) when (Now - Time) >= Period -> false;
in_period(_, _, _) -> true.

%% -- FUSE MANIPULATION ------------------
blow(#fuse { enabled = false }) -> ok;
blow(#fuse { name = Name }) ->
    ets:insert(?TAB, {Name, blown}),
    fuse_event:notify({Name, blown}).

fix(#fuse { enabled = false }) -> ok;
fix(#fuse { name = Name, ty = TY }) ->
    ets:insert(?TAB, {Name, TY}),
    fuse_event:notify({Name, TY}),
    ok.

delete(#fuse { name = Name }) ->
    ets:delete(?TAB, Name),
    fuse_event:notify({Name, removed}),
    ok.

%% -- TIMERS ----
install_timer(#fuse { enabled = false } = F) -> F;
install_timer(#fuse { name = Name, timer_ref = none, heal_time = HealTime } = F) ->
    TRef = fuse_time:send_after(HealTime, self(), {reset, Name}),
    F#fuse{ timer_ref = TRef };
install_timer(F) ->
    F.

reset_timer(#fuse { timer_ref = none } = F) -> F;
reset_timer(#fuse { timer_ref = TRef } = F) ->
    _ = fuse_time:cancel_timer(TRef),
    F#fuse { timer_ref = none }.

install_metrics(#fuse { name = N }) ->
    StatsPlugin = application:get_env(fuse, stats_plugin, fuse_stats_ets),
    _ = StatsPlugin:init(N),
    ok.