%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- % ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et nomod: %%% %%%------------------------------------------------------------------------ %%% @doc %%% ==CloudI Router Service== %%% @end %%% %%% MIT License %%% %%% Copyright (c) 2014-2022 Michael Truog %%% %%% Permission is hereby granted, free of charge, to any person obtaining a %%% copy of this software and associated documentation files (the "Software"), %%% to deal in the Software without restriction, including without limitation %%% the rights to use, copy, modify, merge, publish, distribute, sublicense, %%% and/or sell copies of the Software, and to permit persons to whom the %%% Software is furnished to do so, subject to the following conditions: %%% %%% The above copyright notice and this permission notice shall be included in %%% all copies or substantial portions of the Software. %%% %%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR %%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, %%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE %%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER %%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING %%% FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER %%% DEALINGS IN THE SOFTWARE. %%% %%% @author Michael Truog %%% @copyright 2014-2022 Michael Truog %%% @version 2.0.5 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_service_router). -author('mjtruog at protonmail dot com'). -behaviour(cloudi_service). %% external interface %% cloudi_service callbacks -export([cloudi_service_init/4, cloudi_service_handle_request/11, cloudi_service_handle_info/3, cloudi_service_terminate/3]). -include_lib("cloudi_core/include/cloudi_logger.hrl"). -include_lib("cloudi_core/include/cloudi_service.hrl"). -define(DEFAULT_SSH, undefined). % Enable a ssh server for receiving remote service requests. % The user_dir and system_dir options are required options. % The ip option and port option will likely need to be set % based on your network configuration. % % Example setup from scratch may be like: % $ ssh-keygen -t rsa -f system_dir/ssh_host_rsa_key % $ ssh-keygen -t rsa -f user_dir/id_rsa % $ cat user_dir/id_rsa.pub >> user_dir/authorized_keys % $ chmod 600 user_dir/authorized_keys % % (user_dir could be the path to ${HOME}/.ssh and % system_dir could be /etc/ssh % if permissions and configuration files are correct there) -define(DEFAULT_ADD_PREFIX, true). % The service configuration prefix is automatically added % as a prefix for each of the destinations (all service_names strings). -define(DEFAULT_PARAMETERS_ALLOWED_DEFAULT, true). % Parameters (strings matching the wildcard characters * and ?) % are allowed to be used for constructing destination service names. -define(DEFAULT_PARAMETERS_STRICT_MATCHING_DEFAULT, true). % All * and ? wildcard characters must be used to create exact % service names. That means the parameters_selected list must have % indexes to select each parameter % (each wildcard character in the incoming service name pattern). -define(DEFAULT_VALIDATE_REQUEST_INFO, undefined). -define(DEFAULT_VALIDATE_REQUEST, undefined). -define(DEFAULT_FAILURES_SOURCE_DIE, false). -define(DEFAULT_FAILURES_SOURCE_MAX_COUNT, 2). % see below: % (similar to the MaxR configuration value for services) -define(DEFAULT_FAILURES_SOURCE_MAX_PERIOD, 60). % seconds, see below: % (similar to the MaxT configuration value for services) % If you want the source service to eventually fail, % use the service's MaxT/MaxR as the failures_source_max_period value % (e.g., 300/5 == 60 seconds). Can also use the value 'infinity' % to accumulate a failure count indefinitely. % destinations configuration arguments -define(DEFAULT_REMOTE, undefined). % Should the service request be routed to a remote host? % Takes a list of options, with only the host_name option required. % Defaults are taken from the ssh server configuration, % if it was configured. -define(DEFAULT_MODE, round_robin). % Load-balance among separate service names with % 'round_robin' order or 'random' order. % DEFAULT_PARAMETERS_ALLOWED is set based on % DEFAULT_PARAMETERS_ALLOWED_DEFAULT % DEFAULT_PARAMETERS_STRICT_MATCHING is set based on % DEFAULT_PARAMETERS_STRICT_MATCHING_DEFAULT -define(DEFAULT_PARAMETERS_SELECTED, []). % A list of parameter indexes (1-based index) to be used when % constructing destination service names. That means that the first % instance of * or ? is index 1. If [2] was provided, the second % instance of * or ? would be used to substitute the first * or ? % in all the destination service_names strings. -define(DEFAULT_HTTP_REDIRECT, undefined). % A HTTP 301 redirect response will be provided. % The incoming service request may have been sent from the % HTTP/HTTPS server services cloudi_service_http_cowboy or % cloudi_service_http_elli. The URL string provided may have parameter % substitution but will not have a prefix added due to add_prefix. % The destination URL string will be provided as the % "Location" HTTP header value. If http_redirect is used, % remote, mode and service_names are not used. -define(DEFAULT_SERVICE_NAMES, []). % The destination service names to be used for forwarding % service requests. If * or ? wildcard characters are used % in the service name strings, parameters_allowed needs to be true. % The destination service name strings may have consecutive % wildcard characters for substitution. -record(destination_forward, { parameters_allowed :: boolean(), parameters_strict_matching :: boolean(), parameters_selected :: list(pos_integer()), remote :: undefined | cloudi_service_router_client:state(), index = 1 :: pos_integer(), length :: pos_integer(), mode :: random | round_robin, service_names :: tuple() }). -record(destination_return, { parameters_allowed :: boolean(), parameters_strict_matching :: boolean(), parameters_selected :: list(pos_integer()), http_redirect :: nonempty_string() }). -record(state, { ssh :: undefined | cloudi_service_router_ssh_server:state(), validate_request_info :: undefined | fun((any()) -> boolean()), validate_request :: undefined | fun((any(), any()) -> boolean()), failures_source_die :: boolean(), failures_source_max_count :: pos_integer(), failures_source_max_period :: infinity | pos_integer(), failures_source = #{} :: #{pid() := list(cloudi_timestamp:seconds_monotonic())}, % pattern -> #destination_forward{} | #destination_return{} destinations :: trie:trie() }). %%%------------------------------------------------------------------------ %%% External interface functions %%%------------------------------------------------------------------------ %%%------------------------------------------------------------------------ %%% Callback functions from cloudi_service %%%------------------------------------------------------------------------ cloudi_service_init(Args, Prefix, _Timeout, Dispatcher) -> Defaults = [ {ssh, ?DEFAULT_SSH}, {add_prefix, ?DEFAULT_ADD_PREFIX}, {parameters_allowed_default, ?DEFAULT_PARAMETERS_ALLOWED_DEFAULT}, {parameters_strict_matching_default, ?DEFAULT_PARAMETERS_STRICT_MATCHING_DEFAULT}, {validate_request_info, ?DEFAULT_VALIDATE_REQUEST_INFO}, {validate_request, ?DEFAULT_VALIDATE_REQUEST}, {failures_source_die, ?DEFAULT_FAILURES_SOURCE_DIE}, {failures_source_max_count, ?DEFAULT_FAILURES_SOURCE_MAX_COUNT}, {failures_source_max_period, ?DEFAULT_FAILURES_SOURCE_MAX_PERIOD}, {destinations, []}], [SSHOptions, AddPrefix, ParametersAllowedDefault, ParametersStrictMatchingDefault, ValidateRequestInfo0, ValidateRequest0, FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, DestinationsL] = cloudi_proplists:take_values(Defaults, Args), Environment = cloudi_environment:lookup(), SSH = cloudi_service_router_ssh_server:new(SSHOptions, Environment, Dispatcher), true = is_boolean(AddPrefix), true = is_boolean(ParametersAllowedDefault), true = is_boolean(ParametersStrictMatchingDefault), ValidateRequestInfo1 = cloudi_args_type: function_optional(ValidateRequestInfo0, 1), ValidateRequest1 = cloudi_args_type: function_optional(ValidateRequest0, 2), true = is_boolean(FailuresSrcDie), true = is_integer(FailuresSrcMaxCount) andalso (FailuresSrcMaxCount > 0), true = (FailuresSrcMaxPeriod =:= infinity) orelse (is_integer(FailuresSrcMaxPeriod) andalso (FailuresSrcMaxPeriod > 0)), true = is_list(DestinationsL), ConfigDefaults = [ {remote, ?DEFAULT_REMOTE}, {mode, ?DEFAULT_MODE}, {parameters_allowed, ParametersAllowedDefault}, {parameters_strict_matching, ParametersStrictMatchingDefault}, {parameters_selected, ?DEFAULT_PARAMETERS_SELECTED}, {http_redirect, ?DEFAULT_HTTP_REDIRECT}, {service_names, ?DEFAULT_SERVICE_NAMES}], Destinations = lists:foldl(fun({PatternSuffix, L}, D) -> true = is_tuple(hd(L)), cloudi_service:subscribe(Dispatcher, PatternSuffix), [RemoteOptions, Mode, ParametersAllowed, ParametersStrictMatching, ParametersSelected, HttpRedirect, Names0] = cloudi_proplists:take_values(ConfigDefaults, L), Remote = cloudi_service_router_client:new(RemoteOptions, Environment, SSH), true = is_atom(Mode) andalso ((Mode =:= random) orelse (Mode =:= round_robin)), true = is_boolean(ParametersAllowed), true = is_boolean(ParametersStrictMatching), true = is_list(ParametersSelected), true = ((ParametersSelected == []) orelse ((ParametersSelected /= []) andalso (ParametersAllowed =:= true))), true = lists:all(fun(I) -> is_integer(I) andalso I > 0 end, ParametersSelected), true = is_list(Names0), Length = erlang:length(Names0), Destination = if HttpRedirect =:= undefined -> true = (Length > 0), NamesN = if AddPrefix =:= true -> [Prefix ++ Suffix || Suffix <- Names0]; AddPrefix =:= false -> Names0 end, true = lists:all(fun trie:is_bytestring_nonempty/1, NamesN), #destination_forward{parameters_allowed = ParametersAllowed, parameters_strict_matching = ParametersStrictMatching, parameters_selected = ParametersSelected, remote = Remote, length = Length, mode = Mode, service_names = erlang:list_to_tuple(NamesN)}; is_integer(hd(HttpRedirect)) -> undefined = Remote, true = (Length == 0), true = trie:is_bytestring_nonempty(HttpRedirect), #destination_return{parameters_allowed = ParametersAllowed, parameters_strict_matching = ParametersStrictMatching, parameters_selected = ParametersSelected, http_redirect = HttpRedirect} end, trie:store(Prefix ++ PatternSuffix, Destination, D) end, trie:new(), DestinationsL), {ok, #state{ssh = SSH, validate_request_info = ValidateRequestInfo1, validate_request = ValidateRequest1, failures_source_die = FailuresSrcDie, failures_source_max_count = FailuresSrcMaxCount, failures_source_max_period = FailuresSrcMaxPeriod, destinations = Destinations}}. cloudi_service_handle_request(RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, #state{validate_request_info = RequestInfoF, validate_request = RequestF, destinations = Destinations} = State, _Dispatcher) -> case validate(RequestInfoF, RequestF, RequestInfo, Request) of true -> case trie:find(Pattern, Destinations) of {ok, #destination_forward{} = Destination} -> destination_forward(RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, Destination, State); {ok, #destination_return{} = Destination} -> destination_return(RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, Destination, State); error -> request_failed(Source, State) end; false -> request_failed(Source, State) end. cloudi_service_handle_info({'DOWN', _MonitorRef, process, Pid, _Info}, #state{failures_source_die = FailuresSrcDie, failures_source = FailuresSrc} = State, _Dispatcher) -> FailuresSrcNew = if FailuresSrcDie =:= true -> maps:remove(Pid, FailuresSrc); FailuresSrcDie =:= false -> FailuresSrc end, {noreply, State#state{failures_source = FailuresSrcNew}}; cloudi_service_handle_info(Request, State, _Dispatcher) -> {stop, cloudi_string:format("Unknown info \"~w\"", [Request]), State}. cloudi_service_terminate(_Reason, _Timeout, undefined) -> ok; cloudi_service_terminate(_Reason, _Timeout, #state{ssh = SSH}) -> ok = cloudi_service_router_ssh_server:destroy(SSH), ok. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ destination_forward(RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, Destination, #state{destinations = Destinations} = State) -> {NameNext, DestinationNew} = destination_pick(Destination), Parameters = cloudi_service_name:parse(Name, Pattern), case name_parameters(NameNext, Parameters, DestinationNew) of {ok, NameNew} -> DestinationsNew = trie: store(Pattern, DestinationNew, Destinations), forward(RequestType, Name, Pattern, NameNew, RequestInfo, Request, Timeout, Priority, TransId, Source, DestinationNew, State#state{ destinations = DestinationsNew}); {error, Reason} -> ?LOG_ERROR("(~p -> ~p) error: ~p", [Name, NameNext, Reason]), request_failed(Source, State) end. destination_return(_RequestType, Name, Pattern, _RequestInfo, _Request, _Timeout, _Priority, _TransId, Source, #destination_return{ http_redirect = HttpRedirect} = Destination, State) -> Parameters = cloudi_service_name:parse(Name, Pattern), case name_parameters(HttpRedirect, Parameters, Destination) of {ok, HttpRedirectNew} -> {reply, [{<<"status">>, <<"301">>}, {<<"location">>, erlang:list_to_binary(HttpRedirectNew)}], <<>>, State}; {error, Reason} -> ?LOG_ERROR("(~p http_redirect ~p) error: ~p", [Name, HttpRedirect, Reason]), request_failed(Source, State) end. destination_pick(#destination_forward{service_names = {Name}} = Destination) -> {Name, Destination}; destination_pick(#destination_forward{index = Index, length = Length, mode = round_robin, service_names = Names} = Destination) -> IndexNew = if Index == Length -> 1; true -> Index + 1 end, Name = element(Index, Names), {Name, Destination#destination_forward{index = IndexNew}}; destination_pick(#destination_forward{length = Length, mode = random, service_names = Names} = Destination) -> Index = quickrand:uniform(Length), Name = element(Index, Names), {Name, Destination}. name_parameters(Name, [], _) -> {ok, Name}; name_parameters(_, error, _) -> % a mismatch between trie:find_match2/2 and trie:pattern2_parse/2 % functionality should be impossible, but is handled here {error, parameters_invalid}; name_parameters(_, [_ | _], #destination_forward{parameters_allowed = false}) -> {error, parameters_not_allowed}; name_parameters(_, [_ | _], #destination_return{parameters_allowed = false}) -> {error, parameters_not_allowed}; name_parameters(Pattern, Parameters, #destination_forward{parameters_strict_matching = ParametersStrictMatching, parameters_selected = ParametersSelected}) -> cloudi_service_name:new(Pattern, Parameters, ParametersSelected, ParametersStrictMatching); name_parameters(Pattern, Parameters, #destination_return{parameters_strict_matching = ParametersStrictMatching, parameters_selected = ParametersSelected}) -> cloudi_service_name:new(Pattern, Parameters, ParametersSelected, ParametersStrictMatching). forward(_, _, _, NameNew, RequestInfo, Request, Timeout, Priority, _, _, #destination_forward{remote = undefined}, State) -> {forward, NameNew, RequestInfo, Request, Timeout, Priority, State}; forward(RequestType, Name, Pattern, NameNew, RequestInfo, Request, Timeout, Priority, TransId, Source, #destination_forward{remote = Remote}, State) -> Forward = cloudi_service_router_client: forward(RequestType, Name, Pattern, NameNew, RequestInfo, Request, Timeout, Priority, TransId, Source, Remote), if Forward =:= ok -> {noreply, State}; Forward =:= timeout -> {reply, <<>>, State} end. validate_f_return(Value) when is_boolean(Value) -> Value. validate(undefined, undefined, _, _) -> true; validate(undefined, RF, RInfo, R) -> validate_f_return(RF(RInfo, R)); validate(RInfoF, undefined, RInfo, _) -> validate_f_return(RInfoF(RInfo)); validate(RInfoF, RF, RInfo, R) -> validate_f_return(RInfoF(RInfo)) andalso validate_f_return(RF(RInfo, R)). request_failed(Source, #state{failures_source_die = FailuresSrcDie, failures_source_max_count = FailuresSrcMaxCount, failures_source_max_period = FailuresSrcMaxPeriod, failures_source = FailuresSrc} = State) -> {DeadSrc, FailuresSrcNew} = failure(FailuresSrcDie, FailuresSrcMaxCount, FailuresSrcMaxPeriod, Source, FailuresSrc), if DeadSrc =:= true -> {noreply, State#state{failures_source = FailuresSrcNew}}; DeadSrc =:= false -> {reply, <<>>, State#state{failures_source = FailuresSrcNew}} end. failure(false, _, _, _, Failures) -> {false, Failures}; failure(true, MaxCount, MaxPeriod, Pid, Failures) -> case erlang:is_process_alive(Pid) of true -> SecondsNow = cloudi_timestamp:seconds_monotonic(), case maps:find(Pid, Failures) of {ok, FailureList} -> failure_check(SecondsNow, FailureList, MaxCount, MaxPeriod, Pid, Failures); error -> erlang:monitor(process, Pid), failure_check(SecondsNow, [], MaxCount, MaxPeriod, Pid, Failures) end; false -> {true, Failures} end. failure_store(FailureList, FailureCount, MaxCount, Pid, Failures) -> FailuresNew = maps:put(Pid, FailureList, Failures), if FailureCount == MaxCount -> failure_kill(Pid), {true, FailuresNew}; true -> {false, FailuresNew} end. failure_check(SecondsNow, FailureList, MaxCount, infinity, Pid, Failures) -> FailureCountNew = erlang:length(FailureList), failure_store([SecondsNow | FailureList], FailureCountNew + 1, MaxCount, Pid, Failures); failure_check(SecondsNow, FailureList, MaxCount, MaxPeriod, Pid, Failures) -> {FailureCountNew, FailureListNew} = cloudi_timestamp:seconds_filter_monotonic(FailureList, SecondsNow, MaxPeriod), failure_store([SecondsNow | FailureListNew], FailureCountNew + 1, MaxCount, Pid, Failures). failure_kill(Pid) -> erlang:exit(Pid, cloudi_service_router).