kuberlnetes

%%%-------------------------------------------------------------------
%%% @author bnjm
%%% @copyright (C) 2023, leftshift.one software gmbh
%%% @doc
%%% gen_server process providing kubernetes watch functionality. Supervised
%%% by kuberlnetes_watch_sup - each process represents an open watch for a
%%% given resource.
%%% @end
%%% Created : 13. Apr 2023
%%%-------------------------------------------------------------------
-module(kuberlnetes_watch).

-author("bnjm").

-behaviour(gen_server).

-include("kuberlnetes.hrl").

%% api
-export([
    start_link/3,
    stop/1,
    watch/2
]).

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

-type watch_opts() :: #{
    %% process that receives messages generated by the watch
    receiver => atom() | pid(),
    %% decides if the receiver process should be monitored by the watch
    %% should be false if used in supervision trees
    monitor => boolean(),
    supervised => boolean(),
    name => string(),
    namespace => string(),
    kind => supported_kind(),
    server => kuberlnetes:server()
}.
-export_type([watch_opts/0]).

%% currently known to work kinds
-type supported_kind() :: string().

-include_lib("kernel/include/logger.hrl").

-define(DEFAULT_CONNECTION_TIMEOUT, 5000).

-record(state, {
    %% the owning process of this watch
    %% if the owner goes down the watch will close
    receiver :: pid(),
    receiver_monitor_ref :: reference(),
    %% gun connection process
    conn_pid :: pid(),
    %% timer used to indicate and inital connection timeout
    conn_timeout_timer_ref :: reference() | undefined,
    %% last resource version as indicated by bookmarks
    last_resource_version :: binary() | undefined,
    server :: kuberlnetes:server(),
    api_group :: string(),
    kind :: string(),
    name :: string() | undefined,
    namespace :: string(),
    buffer :: binary()
}).

%% ==================================================================
%% api
%% ==================================================================
-spec start_link(Receiver, Server, Opts) -> Result when
    Receiver :: pid() | atom(),
    Server :: kuberlnetes:server(),
    Opts :: watch_opts(),
    Result :: gen_server:start_ret().
start_link(Receiver, Server, Opts) ->
    gen_server:start_link(?MODULE, [Receiver, Server, Opts], []).

-spec stop(Pid) -> Result when
    Pid :: pid(),
    Result :: ok.
stop(Pid) ->
    gen_server:stop(Pid).

-spec watch(Server, Opts) -> Result when
    Server :: kuberlnetes:server(),
    Opts :: kuberlnetes_watch:watch_opts(),
    Result :: {ok, pid()}.
watch(Server, #{supervised := false} = Opts) ->
    start_link(self(), Server, Opts);
watch(Server, #{supervised := false, receiver := Receiver} = Opts) ->
    start_link(Receiver, Server, Opts);
watch(Server, #{receiver := Receiver} = Opts) ->
    supervisor:start_child(kuberlnetes_watch_sup, [Receiver, Server, Opts]);
watch(Server, Opts) ->
    supervisor:start_child(kuberlnetes_watch_sup, [self(), Server, Opts]).

%% ==================================================================
%% callbacks
%% ==================================================================
init([Receiver, Server, Opts]) ->
    {ok, init_state(Receiver, Server, Opts), {continue, open_connection}}.

init_state(Receiver, Server, Opts) ->
    MaybeMonitorRef = maybe_monitor(Receiver, Opts),
    #state{
        receiver = Receiver,
        receiver_monitor_ref = MaybeMonitorRef,
        server = Server,
        api_group = maps:get(api_group, Opts),
        kind = maps:get(kind, Opts),
        namespace = maps:get(namespace, Opts, "default"),
        name = maps:get(name, Opts),
        buffer = <<>>
    }.

handle_continue(open_connection, #state{server = Server} = State) ->
    open_connection(Server),
    TimerRef = erlang:send_after(?DEFAULT_CONNECTION_TIMEOUT, self(), connection_timeout),
    %% once the connection is open a gun_up is received otherwise the timer hits
    {noreply, State#state{conn_timeout_timer_ref = TimerRef}}.

handle_info({gun_up, ConnPid, http}, #state{conn_timeout_timer_ref = TimerRef} = State) ->
    maybe_cancel_timer(TimerRef),
    %% if the gun connection process goes down
    %% simply let the watch crash
    erlang:monitor(process, ConnPid),
    watch_req(ConnPid, State),
    {noreply, State#state{conn_pid = ConnPid}};
handle_info(connection_timeout, State) ->
    %% inform owner and stop normally
    State#state.receiver ! {kuberlnetes_watch_error, connection_timeout},
    {stop, normal, State};
handle_info({gun_data, _ConnPid, _Ref, nofin, Data}, #state{buffer = Buffer} = State) when
    binary_part(Data, {byte_size(Data), -1}) =:= <<"\n">>
->
    M = jsone:decode(<<Buffer/binary, Data/binary>>),
    NewState = handle_data(M, State),
    {noreply, NewState#state{buffer = <<>>}};
handle_info({gun_data, _ConnPid, _Ref, nofin, Data}, #state{buffer = Buffer} = State) ->
    NewState = State#state{buffer = <<Buffer/binary, Data/binary>>},
    {noreply, NewState};
handle_info(
    {gun_data, ConnPid, _Ref, fin, Data}, #state{last_resource_version = ResourceVersion} = State
) ->
    %% chunked transfer ended - restart from latest bookmark if the received
    %% message is not a "gone" message
    case is_gone_message(Data) of
        true ->
            ?LOG_WARNING(#{
                event => "kuberlnetes_watch_gone",
                info => #{
                    "message" => "Current bookmark does not match latest resource version",
                    "bookmark" => ResourceVersion,
                    "action" => reset_cache
                }
            }),
            NewState = State#state{last_resource_version = undefined},
            watch_req(ConnPid, NewState),
            {noreply, NewState};
        false ->
            watch_req(ConnPid, State),
            {noreply, State}
    end;
handle_info({gun_response, _ConnPid, _Ref, nofin, 200, Headers}, State) ->
    case proplists:get_value(<<"transfer-encoding">>, Headers) of
        <<"chunked">> ->
            {noreply, State};
        _ ->
            State#state.receiver ! {kuberlnetes_watch_error, not_chunked},
            {stop, normal, State}
    end;
handle_info({gun_response, _ConnPid, _Ref, _, Status, _}, State) ->
    ?LOG_ERROR(#{
        error => "kuberlnetes_watch_invalid_resp",
        info =>
            #{
                "message" => "Unexpected response from API server",
                "status" => Status
            }
    }),
    State#state.receiver ! {kuberlnetes_watch_error, invalid_response},
    {stop, normal, State};
handle_info({gun_down, _ConnPif, http, closed, _}, State) ->
    %% the api server apperently closed the connection
    %% terminate and restart
    ?LOG_WARNING(#{
        event => "kuberlnetes_watch_close",
        info =>
            #{"reason" => "server_closed_connection"}
    }),
    {stop, connection_closed, State};
handle_info({'DOWN', _Ref, process, ConnPid, _Reason}, #state{conn_pid = ConnPid} = State) ->
    ?LOG_WARNING(#{
        event => "kuberlnetes_watch_connection_died",
        info =>
            #{"reason" => "gun_client_exited"}
    }),
    {stop, connection_died, State};
handle_info({'DOWN', _Ref, process, Owner, _Reason}, #state{receiver = Owner} = State) ->
    ?LOG_WARNING(#{
        event => "kuberlnetes_watch_owner_died",
        info => #{"pid" => Owner, "action" => "closing"}
    }),
    {stop, normal, State};
handle_info(Message, State) ->
    ?LOG_WARNING(#{event => "unhandled_info", info => #{"message" => Message}}),
    {noreply, State}.

handle_call(Message, _From, State) ->
    ?LOG_WARNING(#{event => "unhandled_call", info => #{"message" => Message}}),
    {reply, ok, State}.

handle_cast(Message, State) ->
    ?LOG_WARNING(#{event => "unhandled_cast", info => #{"message" => Message}}),
    {noreply, State}.

terminate(Reason, #state{conn_pid = ConnPid, receiver_monitor_ref = OwnerRef}) ->
    ?LOG_WARNING(#{event => "terminated", info => #{"reason" => Reason}}),
    gun:close(ConnPid),
    OwnerRef =/= undefined andalso erlang:demonitor(OwnerRef),
    ok.

%% ==================================================================
%% private parts
%% ==================================================================
handle_data(#{<<"type">> := <<"BOOKMARK">>, <<"object">> := Obj}, State) ->
    ?LOG_DEBUG(#{event => "kuberlnetes_watch_bookmark_recv"}),
    #{<<"metadata">> := #{<<"resourceVersion">> := Rv}} = Obj,
    State#state{last_resource_version = Rv};
handle_data(#{<<"type">> := T} = Data, #state{receiver = Owner} = State) ->
    TypeAtom = to_atom_type(T),
    #{<<"object">> := O} = Data,
    Owner ! {kuberlnetes_watch, TypeAtom, O},
    State;
handle_data(_, _) ->
    error(invalid_response).

is_gone_message(<<>>) ->
    false;
is_gone_message(Binary) when is_binary(Binary) ->
    case jsone:try_decode(Binary) of
        {ok, Map, _} ->
            is_gone_message(Map);
        {ok, Map} ->
            is_gone_message(Map);
        {error, R} ->
            ?LOG_WARNING(#{
                event => "kuberlnetes_unprocessable_entity",
                info => #{
                    entity => Binary,
                    reason => R
                }
            }),
            false;
        R ->
            ?LOG_WARNING(#{
                event => "kuberlnetes_unknown_jsone_response",
                info => #{
                    entity => Binary,
                    reponse => R
                }
            }),
            false
    end;
is_gone_message(#{<<"type">> := <<"ERROR">>, <<"object">> := #{<<"code">> := 410}}) ->
    true;
is_gone_message(_) ->
    false.

to_atom_type(<<"ADDED">>) -> added;
to_atom_type(<<"MODIFIED">>) -> modified;
to_atom_type(<<"DELETED">>) -> deleted;
to_atom_type(<<"ERROR">>) -> error.

open_connection(Server) ->
    ?LOG_DEBUG(#{
        event => "connecting", info => #{"host" => Server#server.host, "port" => Server#server.port}
    }),
    {ok, ConnPid} = gun:open(Server#server.host, Server#server.port, #{
        % transport => tls,
        protocols => [http],
        http_opts => #{keepalive => 30000},
        tls_opts => [
            {verify, verify_none},
            {cacerts, [Server#server.ca_cert]}
        ]
    }),
    ConnPid.
watch_req(ConnPid, State) ->
    gun:get(
        ConnPid,
        watch_path(State),
        kuberlnetes:headers(State#state.server, binary)
    ).

maybe_cancel_timer(undefined) ->
    false;
maybe_cancel_timer(TimerRef) when is_reference(TimerRef) ->
    erlang:cancel_timer(TimerRef);
maybe_cancel_timer(_) ->
    error(unexpeced_timer_ref).

maybe_monitor(_, #{monitor := false}) ->
    undefined;
maybe_monitor(Receiver, #{monitor := true}) ->
    monitor(process, Receiver);
maybe_monitor(Receiver, Opts) ->
    maybe_monitor(Receiver, Opts#{monitor => true}).

watch_path(#state{last_resource_version = undefined, name = undefined} = State) ->
    io_lib:format(
        "~s/namespaces/~s/~s?allowWatchBookmarks&watch=true",
        [
            State#state.api_group,
            State#state.namespace,
            State#state.kind
        ]
    );
watch_path(#state{last_resource_version = Rv, name = undefined} = State) ->
    io_lib:format(
        "~s/namespaces/~s/~s?allowWatchBookmarks&resourceVersion=~s&watch=true",
        [
            State#state.api_group,
            State#state.namespace,
            State#state.kind,
            Rv
        ]
    );
watch_path(#state{last_resource_version = undefined} = State) ->
    io_lib:format(
        "~s/namespaces/~s/~s?fieldSelector=metadata.name%3D~s&allowWatchBookmarks&watch=true",
        [
            State#state.api_group,
            State#state.namespace,
            State#state.kind,
            State#state.name
        ]
    );
watch_path(#state{last_resource_version = Rv} = State) ->
    io_lib:format(
        "~s/namespaces/~s/~s?fieldSelector=metadata.name%3D~s&resourceVersion=~s&allowWatchBookmarks&watch=true",
        [
            State#state.api_group,
            State#state.namespace,
            State#state.kind,
            State#state.name,
            Rv
        ]
    ).