diff --git a/src/eradius_client_mngr.erl b/src/eradius_client_mngr.erl index 83e6e12f..c905743c 100644 --- a/src/eradius_client_mngr.erl +++ b/src/eradius_client_mngr.erl @@ -39,9 +39,15 @@ Pool-based failover in `m:eradius_client` automatically skips unreachable server == Socket pool == -The client opens `no_ports` UDP sockets (default: 1) on OS-assigned ports. -Each socket supports up to 256 concurrent requests (one per RADIUS request id). -Increase `no_ports` for higher concurrency requirements. +For each RADIUS server the client maintains a dynamic pool of connected UDP sockets +on OS-assigned source ports. Up to `no_ports` (default: 10) sockets are actively +issuing request ids at once; each socket issues request ids 0..255 exactly once, then +is retired — held open for `reqid_reuse_timeout` ms (default: 30000) so its source port +is not reused while the server's duplicate-detection window is still open — and finally +closed. Reusing a request id therefore always happens on a fresh source port, which the +server sees as a distinct client. The total number of sockets per server is bounded by +`max_ports_per_server` (default: 256); once reached, `wanna_send` returns `{error, no_ports}` +so the caller can apply backpressure. """. -behaviour(gen_server). @@ -58,8 +64,8 @@ Increase `no_ports` for higher concurrency requirements. -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -ifdef(TEST). --export([get_state/1, servers/1, server/2, get_socket_count/1]). --ignore_xref([get_state/1, servers/1, server/2, get_socket_count/1]). +-export([get_state/1, servers/1, server/2]). +-ignore_xref([get_state/1, servers/1, server/2]). -endif. -ignore_xref([start_client/1, start_client/2]). @@ -102,7 +108,9 @@ Increase `no_ports` for higher concurrency requirements. inet_backend => inet | socket, ip => any | inet:ip_address(), active_n => once | non_neg_integer(), - no_ports => non_neg_integer(), + no_ports => pos_integer(), + max_ports_per_server => pos_integer(), + reqid_reuse_timeout => pos_integer(), recbuf => non_neg_integer(), sndbuf => non_neg_integer(), metrics_callback => eradius_req:metrics_callback() @@ -115,7 +123,9 @@ Increase `no_ports` for higher concurrency requirements. family := inet | inet6, ip := any | inet:ip_address(), active_n := once | non_neg_integer(), - no_ports := non_neg_integer(), + no_ports := pos_integer(), + max_ports_per_server => pos_integer(), + reqid_reuse_timeout => pos_integer(), recbuf := non_neg_integer(), sndbuf := non_neg_integer(), metrics_callback := 'undefined' | eradius_req:metrics_callback() @@ -125,6 +135,13 @@ Increase `no_ports` for higher concurrency requirements. -export_type([server_name/0, server_pool/0, servers/0, client_opts/0]). +-define(RECONFIGURE_TIMEOUT, 15000). +-define(DEFAULT_MAX_RETRIES, 20). +-define(DEFAULT_DOWN_TIME, 1000). +-define(DEFAULT_K_PORTS, 10). +-define(DEFAULT_MAX_PORTS_PER_SERVER, 256). +-define(DEFAULT_REQID_REUSE_TIMEOUT, 30000). + -record(state, { owner :: pid(), config :: client_config(), @@ -132,15 +149,19 @@ Increase `no_ports` for higher concurrency requirements. client_addr :: any | inet:ip_address(), servers :: servers(), socket_id :: {Family :: inet | inet6, IP :: any | inet:ip_address()}, - no_ports = 1 :: pos_integer(), - idcounters = maps:new() :: map(), - sockets = array:new() :: array:array(), + k_ports = ?DEFAULT_K_PORTS :: pos_integer(), + max_ports = ?DEFAULT_MAX_PORTS_PER_SERVER :: pos_integer(), + reqid_reuse_timeout = ?DEFAULT_REQID_REUSE_TIMEOUT :: pos_integer(), + pools = #{} :: #{server_addr() => pool()}, + socket_refs = #{} :: #{reference() => server_addr()}, + no_ports_rejections = 0 :: non_neg_integer(), metrics_callback :: undefined | eradius_req:metrics_callback() }). --define(RECONFIGURE_TIMEOUT, 15000). --define(DEFAULT_MAX_RETRIES, 20). --define(DEFAULT_DOWN_TIME, 1000). +-type server_addr() :: {inet:ip_address(), inet:port_number()}. +-type filler() :: #{pid := pid(), monitor := reference(), + next_id := 0..255, issued := 0..256}. +-type pool() :: #{active := queue:queue(filler()), cooling := [pid()]}. %%%========================================================================= %%% API @@ -214,12 +235,6 @@ get_state(ServerRef) -> Values = tl(tuple_to_list(State)), maps:from_list(lists:zip(Keys, Values)). -get_socket_count(ServerRef) -> - #state{owner = Owner} = sys:get_state(ServerRef), - {ok, SockSup} = eradius_client_sup:socket_supervisor(Owner), - Counts = supervisor:count_children(SockSup), - proplists:get_value(active, Counts). - servers(ServerRef) -> #state{servers = Servers} = sys:get_state(ServerRef), maps:fold( @@ -257,33 +272,35 @@ init([Owner, #{name := ClientName, servers := Servers, config = Config, servers = Servers, socket_id = socket_id(Config), - no_ports = NPorts, + k_ports = NPorts, + max_ports = maps:get(max_ports_per_server, Config, + ?DEFAULT_MAX_PORTS_PER_SERVER), + reqid_reuse_timeout = maps:get(reqid_reuse_timeout, Config, + ?DEFAULT_REQID_REUSE_TIMEOUT), metrics_callback = MetricsCallback }, {ok, State}. %% @private handle_call({wanna_send, Candidates, Tried}, _From, - #state{ - client_name = ClientName, - client_addr = ClientAddr, - servers = Servers, - no_ports = NoPorts, idcounters = IdCounters, - sockets = Sockets, - metrics_callback = MetricsCallback} = State0) -> + #state{client_name = ClientName, client_addr = ClientAddr, + servers = Servers, + metrics_callback = MetricsCallback} = State0) -> case select_server(Candidates, Tried, Servers) of {ok, {ServerName, #{ip := IP, port := Port} = Server}} -> ServerAddr = {IP, Port}, - {PortIdx, ReqId, NewIdCounters} = - next_port_and_req_id(ServerAddr, NoPorts, IdCounters), - {SocketProcess, NewSockets} = find_socket_process(PortIdx, Sockets, State0), - State = State0#state{idcounters = NewIdCounters, sockets = NewSockets}, - ReqInfo = - #{server => ServerName, server_addr => ServerAddr, - client => ClientName, client_addr => ClientAddr, - metrics_callback => MetricsCallback}, - Reply = {ok, {SocketProcess, ReqId, ServerName, Server, ReqInfo}}, - {reply, Reply, State}; + case allocate(ServerAddr, State0) of + {ok, Pid, ReqId, State} -> + ReqInfo = + #{server => ServerName, server_addr => ServerAddr, + client => ClientName, client_addr => ClientAddr, + metrics_callback => MetricsCallback}, + {reply, {ok, {Pid, ReqId, ServerName, Server, ReqInfo}}, State}; + {error, no_ports, #state{no_ports_rejections = N} = State1} -> + State = State1#state{no_ports_rejections = N + 1}, + ?LOG(warning, "RADIUS client port pool for ~p saturated", [ServerAddr]), + {reply, {error, no_ports}, State} + end; {error, _} = Error -> {reply, Error, State0} end; @@ -304,8 +321,14 @@ handle_call({failed, _Peer}, _From, State) -> handle_call({reconfigure, Opts}, _From, #state{config = OConfig} = State0) -> case client_config(maps:merge(OConfig, Opts)) of {ok, #{servers := Servers} = Config} -> - State1 = State0#state{config = Config, servers = Servers}, - State = reconfigure_address(Config, State1), + State1 = State0#state{ + config = Config, servers = Servers, + max_ports = maps:get(max_ports_per_server, Config, + ?DEFAULT_MAX_PORTS_PER_SERVER), + reqid_reuse_timeout = maps:get(reqid_reuse_timeout, Config, + ?DEFAULT_REQID_REUSE_TIMEOUT)}, + State2 = reconfigure_address(Config, State1), + State = drop_removed_pools(Servers, State2), {reply, ok, State}; {error, _} = Error -> @@ -319,6 +342,18 @@ handle_call(_OtherCall, _From, State) -> %% @private handle_cast(_Msg, State) -> {noreply, State}. +%% @private +handle_info({'DOWN', Ref, process, Pid, _Reason}, + #state{socket_refs = Refs} = State) -> + case maps:take(Ref, Refs) of + {ServerAddr, Refs1} -> + Pool = pool_of(ServerAddr, State), + Pool1 = remove_socket(Pid, Ref, Pool), + {noreply, put_pool(ServerAddr, Pool1, State#state{socket_refs = Refs1})}; + error -> + {noreply, State} + end; + %% @private handle_info({timeout, _, {reset, Peer}}, #state{servers = Servers0} = State0) -> Servers = @@ -372,7 +407,9 @@ socket_id_str({_, IP}) when is_atom(IP) -> default_client_opts() -> #{family => inet6, ip => any, - no_ports => 10, + no_ports => ?DEFAULT_K_PORTS, + max_ports_per_server => ?DEFAULT_MAX_PORTS_PER_SERVER, + reqid_reuse_timeout => ?DEFAULT_REQID_REUSE_TIMEOUT, active_n => 100, recbuf => 8192, sndbuf => 131072, @@ -481,76 +518,137 @@ client_config(Opts0) -> end. reconfigure_address(#{no_ports := NPorts} = Config, - #state{socket_id = OAdd, sockets = Sockts} = State) -> + #state{socket_id = OAdd, socket_refs = Refs} = State) -> NAdd = socket_id(Config), case OAdd of - NAdd -> + NAdd -> reconfigure_ports(NPorts, State); _ -> ?LOG(info, "Reopening RADIUS client sockets (client_ip changed to ~s)", [socket_id_str(NAdd)]), - array:map( - fun(_PortIdx, undefined) -> - ok; - (_PortIdx, Socket) -> - eradius_client_socket:close(Socket) - end, Sockts), - Counters = fix_counters(NPorts, State#state.idcounters), - State#state{sockets = array:new(), socket_id = NAdd, - no_ports = NPorts, idcounters = Counters} + close_all_pools(State), + maps:foreach(fun(Ref, _SA) -> erlang:demonitor(Ref, [flush]) end, Refs), + State#state{socket_id = NAdd, k_ports = NPorts, + pools = #{}, socket_refs = #{}} end. -reconfigure_ports(NPorts, #state{no_ports = OPorts, sockets = Sockets} = State) -> - if - OPorts =< NPorts -> - State#state{no_ports = NPorts}; - true -> - Counters = fix_counters(NPorts, State#state.idcounters), - NSockets = close_sockets(NPorts, Sockets), - State#state{sockets = NSockets, no_ports = NPorts, idcounters = Counters} - end. +close_all_pools(#state{pools = Pools}) -> + maps:foreach( + fun(_ServerAddr, Pool) -> + lists:foreach(fun eradius_client_socket:close/1, pool_pids(Pool)) + end, Pools), + ok. -fix_counters(NPorts, Counters) -> - maps:map(fun(_Peer, Value = {NextPortIdx, _NextReqId}) when NextPortIdx < NPorts -> Value; - (_Peer, {_NextPortIdx, NextReqId}) -> {0, NextReqId} - end, Counters). - -close_sockets(NPorts, Sockets) -> - case array:size(Sockets) =< NPorts of - true -> - Sockets; - false -> - List = array:to_list(Sockets), - {_, Rest} = lists:split(NPorts, List), - lists:map( - fun(undefined) -> ok; - (Socket) -> eradius_client_socket:close(Socket) - end, Rest), - array:resize(NPorts, Sockets) +%% Drop pools whose server is no longer configured; close their sockets and +%% demonitor them. +drop_removed_pools(Servers, #state{pools = Pools} = State) -> + Active = sets:from_list( + maps:fold(fun(_, #{ip := IP, port := Port}, Acc) -> [{IP, Port} | Acc]; + (_, _Pool, Acc) -> Acc + end, [], Servers)), + Drop = [SA || SA <- maps:keys(Pools), not sets:is_element(SA, Active)], + lists:foldl(fun drop_one_pool/2, State, Drop). + +drop_one_pool(ServerAddr, #state{pools = Pools, socket_refs = Refs} = State) -> + lists:foreach(fun eradius_client_socket:close/1, + pool_pids(maps:get(ServerAddr, Pools))), + Refs1 = + maps:filter( + fun(Ref, SA) when SA =:= ServerAddr -> erlang:demonitor(Ref, [flush]), false; + (_Ref, _SA) -> true + end, Refs), + State#state{pools = maps:remove(ServerAddr, Pools), socket_refs = Refs1}. + +pool_pids(#{active := Active, cooling := Cooling}) -> + [maps:get(pid, F) || F <- queue:to_list(Active)] ++ Cooling. + +reconfigure_ports(NPorts, State) -> + State#state{k_ports = NPorts}. + +-spec new_pool() -> pool(). +new_pool() -> #{active => queue:new(), cooling => []}. + +pool_of(ServerAddr, #state{pools = Pools}) -> + maps:get(ServerAddr, Pools, new_pool()). + +put_pool(ServerAddr, Pool, #state{pools = Pools} = State) -> + State#state{pools = Pools#{ServerAddr => Pool}}. + +%% active fillers + cooling (retired-but-open) sockets. cooling pids are reclaimed +%% when their socket exits (the 'DOWN' handler), bounding the per-server total. +pool_total(#{active := A, cooling := C}) -> queue:len(A) + length(C). + +%% Drop a dead socket (by pid) from a pool, whether it was an active filler or +%% a cooling socket. +remove_socket(Pid, _Ref, #{active := Active, cooling := Cooling} = Pool) -> + Pool#{active := queue:delete_with(fun(F) -> maps:get(pid, F) =:= Pid end, Active), + cooling := lists:delete(Pid, Cooling)}. + +%% Allocate {Pid, ReqId} for ServerAddr, growing/rolling the pool as needed. +-spec allocate(server_addr(), #state{}) -> + {ok, pid(), 0..255, #state{}} | {error, no_ports, #state{}}. +allocate(ServerAddr, State0) -> + State1 = ensure_active(ServerAddr, State0), + Pool = pool_of(ServerAddr, State1), + case queue:out(maps:get(active, Pool)) of + {empty, _} -> + {error, no_ports, State1}; + {{value, #{pid := Pid, next_id := ReqId, issued := Issued0} = F}, Rest} -> + Issued = Issued0 + 1, + case Issued >= 256 of + true -> + %% filler exhausted: retire it (socket cools+closes itself), + %% drop it from the active queue, and open a replacement. + ok = eradius_client_socket:retire(Pid), + Pool1 = Pool#{active := Rest, + cooling := [Pid | maps:get(cooling, Pool)]}, + State2 = put_pool(ServerAddr, Pool1, State1), + State3 = ensure_active(ServerAddr, State2), + {ok, Pid, ReqId, State3}; + false -> + %% round-robin: advance this filler and re-enqueue it at the tail. + F1 = F#{next_id := (ReqId + 1) rem 256, issued := Issued}, + Pool1 = Pool#{active := queue:in(F1, Rest)}, + {ok, Pid, ReqId, put_pool(ServerAddr, Pool1, State1)} + end end. -next_port_and_req_id(Peer, NumberOfPorts, Counters) -> - case Counters of - #{Peer := {NextPortIdx, ReqId}} when ReqId < 255 -> - NextReqId = (ReqId + 1); - #{Peer := {PortIdx, 255}} -> - NextPortIdx = (PortIdx + 1) rem NumberOfPorts, - NextReqId = 0; - _ -> - NextPortIdx = erlang:phash2(Peer, NumberOfPorts), - NextReqId = 0 - end, - NewCounters = Counters#{Peer => {NextPortIdx, NextReqId}}, - {NextPortIdx, NextReqId, NewCounters}. +%% Open fillers until there are K active (or the per-server cap is hit, or a +%% socket open fails). Idempotent. +-spec ensure_active(server_addr(), #state{}) -> #state{}. +ensure_active(ServerAddr, #state{k_ports = K, max_ports = Max} = State) -> + Pool = pool_of(ServerAddr, State), + Active = maps:get(active, Pool), + case queue:len(Active) < K andalso pool_total(Pool) < Max of + true -> + case open_filler(ServerAddr, State) of + {ok, Filler, State1} -> + Pool1 = pool_of(ServerAddr, State1), + Pool2 = Pool1#{active := queue:in(Filler, maps:get(active, Pool1))}, + ensure_active(ServerAddr, put_pool(ServerAddr, Pool2, State1)); + {error, _} -> + State + end; + false -> + State + end. -find_socket_process(PortIdx, Sockets, #state{owner = Owner, config = Config}) -> - case array:get(PortIdx, Sockets) of - undefined -> - {ok, Supervisor} = eradius_client_sup:socket_supervisor(Owner), - {ok, Socket} = eradius_client_socket:new(Supervisor, Config), - {Socket, array:set(PortIdx, Socket, Sockets)}; - Socket -> - {Socket, Sockets} +-spec open_filler(server_addr(), #state{}) -> + {ok, filler(), #state{}} | {error, term()}. +open_filler(ServerAddr, #state{owner = Owner, config = Config, + reqid_reuse_timeout = RT, + socket_refs = Refs} = State) -> + {ok, Supervisor} = eradius_client_sup:socket_supervisor(Owner), + SockConfig = Config#{server_addr => ServerAddr, reqid_reuse_timeout => RT}, + case eradius_client_socket:new(Supervisor, SockConfig) of + {ok, Pid} -> + Ref = erlang:monitor(process, Pid), + Filler = #{pid => Pid, monitor => Ref, next_id => 0, issued => 0}, + {ok, Filler, State#state{socket_refs = Refs#{Ref => ServerAddr}}}; + {error, _} = Error -> + ?LOG(warning, "could not open RADIUS client socket for ~p: ~p", + [ServerAddr, Error]), + Error end. client_mngr_pid(SupPid) -> diff --git a/src/eradius_client_socket.erl b/src/eradius_client_socket.erl index 4659183d..b8bd4aba 100644 --- a/src/eradius_client_socket.erl +++ b/src/eradius_client_socket.erl @@ -9,14 +9,21 @@ -behaviour(gen_server). %% API --export([new/2, start_link/1, send_request/5, close/1]). +-export([new/2, start_link/1, send_request/5, retire/1, close/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -ignore_xref([start_link/1]). --record(state, {family, socket, active_n, pending, mode, counter}). +-record(state, {family, socket, active_n, pending, mode, + server_addr, cooldown, cooldown_done = false}). + +-define(DEFAULT_REQID_REUSE_TIMEOUT, 30000). + +%% Hard ceiling (multiple of the cooldown) after which a retired socket is +%% force-closed even if a straggler request is still pending, to bound resources. +-define(FORCE_CLOSE_FACTOR, 2). %% Safety margin added to the per-request timeout for the gen_server:call. %% The socket process enforces the real timeout and replies {error,timeout}; @@ -52,6 +59,9 @@ send_request(Socket, Peer, ReqId, Request, Timeout) -> call_timeout(infinity) -> infinity; call_timeout(Timeout) when is_integer(Timeout) -> Timeout + ?CALL_TIMEOUT_MARGIN. +retire(Socket) -> + gen_server:cast(Socket, retire). + close(Socket) -> gen_server:cast(Socket, close). @@ -59,30 +69,47 @@ close(Socket) -> %%% gen_server callbacks %%%=================================================================== -init([#{family := Family, active_n := ActiveN} = Config]) -> +init([#{family := Family, active_n := ActiveN, + server_addr := {SrvIP, SrvPort}} = Config]) -> Opts = inet_opts(Config, [{active, ActiveN}, binary, Family]), {ok, Socket} = gen_udp:open(0, Opts), + case connect(Socket, Family, SrvIP, SrvPort) of + {ok, ConnIP} -> + State = #state{ + family = Family, + socket = Socket, + active_n = ActiveN, + pending = #{}, + mode = active, + server_addr = {ConnIP, SrvPort}, + cooldown = maps:get(reqid_reuse_timeout, Config, + ?DEFAULT_REQID_REUSE_TIMEOUT) + }, + {ok, State}; + {error, Reason} -> + gen_udp:close(Socket), + {stop, Reason} + end. - State = #state{ - family = Family, - socket = Socket, - active_n = ActiveN, - pending = #{}, - mode = active - }, - {ok, State}. - -handle_call({send_request, {IP, Port}, ReqId, Request, Timeout}, From, - #state{family = Family, socket = Socket} = State) -> - case send_ip(Family, IP) of - {ok, SendIP} -> - case gen_udp:send(Socket, SendIP, Port, Request) of +%% Map the server IP into the socket family and connect; close+stop on any error. +connect(Socket, Family, SrvIP, SrvPort) -> + case send_ip(Family, SrvIP) of + {ok, ConnIP} -> + case gen_udp:connect(Socket, ConnIP, SrvPort) of ok -> - ReqKey = {SendIP, Port, ReqId}, - {noreply, pending_request(ReqKey, From, Timeout, State)}; + {ok, ConnIP}; {error, _} = Error -> - {reply, Error, State} + Error end; + {error, _} = Error -> + Error + end. + +handle_call({send_request, _Peer, ReqId, Request, Timeout}, From, + #state{socket = Socket} = State) -> + case gen_udp:send(Socket, Request) of + ok -> + {noreply, pending_request(ReqId, From, Timeout, State)}; {error, _} = Error -> {reply, Error, State} end; @@ -90,6 +117,14 @@ handle_call({send_request, {IP, Port}, ReqId, Request, Timeout}, From, handle_call(_Request, _From, State) -> {reply, {error, unknown_request}, State}. +handle_cast(retire, #state{mode = retiring} = State) -> + %% idempotent: a second retire must not start another timer set + {noreply, State}; +handle_cast(retire, #state{cooldown = Cooldown} = State) -> + erlang:start_timer(Cooldown, self(), cooldown_expired), + erlang:start_timer(Cooldown * ?FORCE_CLOSE_FACTOR, self(), force_close), + noreply_or_stop(State#state{mode = retiring}); + handle_cast(close, #state{pending = Pending} = State) when map_size(Pending) =:= 0 -> {stop, normal, State}; @@ -103,26 +138,33 @@ handle_info({udp_passive, _Socket}, #state{socket = Socket, active_n = ActiveN} inet:setopts(Socket, [{active, ActiveN}]), {noreply, State}; -handle_info({udp, Socket, FromIP, FromPort, Response}, +handle_info({udp, Socket, _FromIP, _FromPort, Response}, State = #state{socket = Socket}) -> flow_control(State), NState = case Response of <> -> <<_, ReqId:8, _/binary>> = Header, - request_done({FromIP, FromPort, ReqId}, {ok, Header, Body}, State); + request_done(ReqId, {ok, Header, Body}, State); _ -> %% discard reply because it was malformed State end, noreply_or_stop(NState); -handle_info({timeout, TRef, ReqKey}, #state{pending = Pending} = State) -> +handle_info({timeout, _TRef, cooldown_expired}, State) -> + noreply_or_stop(State#state{cooldown_done = true}); + +handle_info({timeout, _TRef, force_close}, State) -> + {stop, normal, State}; + +handle_info({timeout, TRef, ReqId}, #state{pending = Pending} = State) + when is_integer(ReqId) -> NState = case Pending of - #{ReqKey := {From, TRef}} -> + #{ReqId := {From, TRef}} -> gen_server:reply(From, {error, timeout}), - State#state{pending = maps:remove(ReqKey, Pending)}; + State#state{pending = maps:remove(ReqId, Pending)}; _ -> State end, @@ -149,20 +191,22 @@ flow_control(_) -> noreply_or_stop(#state{pending = Pending, mode = inactive} = State) when map_size(Pending) =:= 0 -> {stop, normal, State}; +noreply_or_stop(#state{pending = Pending, mode = retiring, cooldown_done = true} = State) + when map_size(Pending) =:= 0 -> + {stop, normal, State}; noreply_or_stop(State) -> {noreply, State}. -pending_request(ReqKey, From, Timeout, - #state{pending = Pending} = State) -> - TRef = erlang:start_timer(Timeout, self(), ReqKey), - State#state{pending = Pending#{ReqKey => {From, TRef}}}. +pending_request(ReqId, From, Timeout, #state{pending = Pending} = State) -> + TRef = erlang:start_timer(Timeout, self(), ReqId), + State#state{pending = Pending#{ReqId => {From, TRef}}}. -request_done(ReqKey, Reply, #state{pending = Pending} = State) -> +request_done(ReqId, Reply, #state{pending = Pending} = State) -> case Pending of - #{ReqKey := {From, TRef}} -> + #{ReqId := {From, TRef}} -> gen_server:reply(From, Reply), erlang:cancel_timer(TRef), - State#state{pending = maps:remove(ReqKey, Pending)}; + State#state{pending = maps:remove(ReqId, Pending)}; _ -> State end. diff --git a/src/eradius_client_socket_sup.erl b/src/eradius_client_socket_sup.erl index db2d5957..fdd03e67 100644 --- a/src/eradius_client_socket_sup.erl +++ b/src/eradius_client_socket_sup.erl @@ -45,9 +45,13 @@ init([]) -> intensity => 5, period => 10}, + %% Pool sockets are managed by eradius_client_mngr: it opens them on demand, + %% monitors them, and reclaims them on exit. They must NOT auto-restart -- a + %% restarted socket would be a pid the manager never learns about (an orphaned + %% FD/source port). The manager reopens a replacement on the next allocation. Child = #{id => eradius_client_socket, start => {eradius_client_socket, start_link, []}, - restart => transient, + restart => temporary, shutdown => 5000, type => worker, modules => [eradius_client_socket]}, diff --git a/test/eradius_client_SUITE.erl b/test/eradius_client_SUITE.erl index 7dbca1e7..9e298990 100644 --- a/test/eradius_client_SUITE.erl +++ b/test/eradius_client_SUITE.erl @@ -59,8 +59,15 @@ common() -> wanna_send, send_request_failover, check_upstream_servers, - no_ports_one_wraps, - clobber_does_not_hang + clobber_does_not_hang, + connected_socket_matches_reply, + retire_holds_then_closes, + retire_waits_for_pending, + client_config_defaults, + pool_rolls_and_retires, + pool_cap_backpressures, + cooling_socket_reclaimed, + rotation_uses_fresh_source_ports ]. -spec groups() -> [ct_suite:ct_group_def(), ...]. @@ -128,6 +135,13 @@ init_per_testcase(check_upstream_servers, Config) -> Config; init_per_testcase(wanna_send, Config) -> start_client(Config), + %% The named client persists across the group; a preceding reconf_address + %% may have left it bound to a non-existent IP. Restore a usable client_ip + %% and the default port count so socket opens succeed. + Family = proplists:get_value(family, Config), + ok = eradius_client_mngr:reconfigure( + ?SERVER, #{ip => eradius_test_lib:localhost(Family, native), + no_ports => 10}), Config; init_per_testcase(_Test, Config) -> Config. @@ -144,135 +158,54 @@ end_per_testcase(check_upstream_servers, Config) -> end_per_testcase(_Test, Config) -> Config. -%% STUFF - -getSocketCount() -> - eradius_client_mngr:get_socket_count(?SERVER). - -testSocket(undefined) -> - true; -testSocket(Pid) -> - not is_process_alive(Pid). - -split(N, List) -> split2(N, [], List). - -split2(0, List1, List2) -> {lists:reverse(List1), List2}; -split2(_, List1, []) -> {lists:reverse(List1), []}; -split2(N, List1, [L|List2]) -> split2(N-1, [L|List1], List2). - -meckStart() -> - ok = meck:new(eradius_client_socket, [passthrough]), - ok = meck:expect(eradius_client_socket, init, - fun(_) -> {ok, undefined} end), - ok = meck:expect(eradius_client_socket, handle_call, - fun(_Request, _From, State) -> {noreply, State} end), - ok = meck:expect(eradius_client_socket, handle_cast, - fun(close, State) -> {stop, normal, State}; - (_Request, State) -> {noreply, State} end), - ok = meck:expect(eradius_client_socket, handle_info, - fun(_Info, State) -> {noreply, State} end), - ok. - -meckStop() -> - ok = meck:unload(eradius_client_socket). - -parse_ip(undefined) -> - {ok, undefined}; -parse_ip(any) -> - {ok, any}; -parse_ip(Address) when is_list(Address) -> - inet_parse:address(Address); -parse_ip(T = {_, _, _, _}) -> - {ok, T}; -parse_ip(T = {_, _, _, _, _, _, _, _}) -> - {ok, T}. - -%% CHECK - -test(true, _Msg) -> true; -test(false, Msg) -> - ct:pal("~s", [Msg]), - false. - -check(OldState, NewState = #{no_ports := P}, null, A) -> check(OldState, NewState, P, A); -check(OldState, NewState = #{socket_id := {_, A}}, P, null) -> check(OldState, NewState, P, A); -check(#{sockets := OS, no_ports := _OP, idcounters := _OC, socket_id := {_, OA}}, - #{sockets := NS, no_ports := NP, idcounters := NC, socket_id := {_, NA}}, - P, A) -> - {ok, PA} = parse_ip(A), - test(PA == NA, "Address not configured") and - case NA of - OA -> - ct:pal("NP: ~p, NC: ~p", [NP, NC]), - {_, Rest} = split(NP, array:to_list(OS)), - test(P == NP,"Ports not configured") and - test(maps:fold( fun(_Peer, {NextPortIdx, _NextReqId}, Akk) -> - Akk and (NextPortIdx =< NP) - end, true, NC), "Invalid port counter") and - test(getSocketCount() =< NP, "Sockets not closed") and - test(array:size(NS) =< NP, "Socket array not resized") and - test(lists:all(fun(Pid) -> testSocket(Pid) end, Rest), "Sockets still available"); - _ -> - test(array:size(NS) == 0, "Socket array not cleaned") and - test(getSocketCount() == 0, "Sockets not closed") and - test(lists:all(fun(Pid) -> testSocket(Pid) end, array:to_list(OS)), "Sockets still available") - end. - %% TESTS send_request(_Config) -> ?equal(accept, eradius_test_handler:send_request(one)), ok. -send(FUN, Ports, Address) -> - meckStart(), - OldState = eradius_client_mngr:get_state(?SERVER), - FUN(), - NewState = eradius_client_mngr:get_state(?SERVER), - true = check(OldState, NewState, Ports, Address), - meckStop(). +%% true iff every pool keeps at most K active fillers (family-agnostic). +all_pools_within_k(St, K) -> + lists:all(fun(#{active := A}) -> queue:len(A) =< K end, + maps:values(maps:get(pools, St))). wanna_send(_Config) -> - lists:map(fun(X) -> - Server = binary_to_atom(<<(X+$A)>>), - FUN = fun() -> eradius_client_mngr:wanna_send(?SERVER, [Server], []) end, - send(FUN, null, null) - end, lists:seq(0, 9)). + %% Allocations stay within K active fillers per server. Server `one` + %% is configured by eradius_test_handler:start_client/2 at port 1812. + K = 10, + lists:foreach( + fun(_) -> + {ok, {_Pid, _Id, one, _S, _I}} = + eradius_client_mngr:wanna_send(?SERVER, [one], []) + end, lists:seq(1, 50)), + St = eradius_client_mngr:get_state(?SERVER), + ?equal(true, all_pools_within_k(St, K)), + ok. reconf_address(Config) -> IP = case proplists:get_value(family, Config) of - ipv4 -> - {7, 13, 23, 42}; - ipv4_mapped_ipv6 -> - inet:ipv4_mapped_ipv6_address({7, 13, 23, 42}); - ipv6 -> - {16#fd96, 16#dcd2, 16#efdb, 16#41c3, 0, 0, 16#100, 1} + ipv4 -> {7, 13, 23, 42}; + ipv4_mapped_ipv6 -> inet:ipv4_mapped_ipv6_address({7, 13, 23, 42}); + ipv6 -> {16#fd96, 16#dcd2, 16#efdb, 16#41c3, 0, 0, 16#100, 1} end, - FUN = fun() -> - eradius_client_mngr:reconfigure(?SERVER, #{ip => IP}), - %% socket shutdown is done asynchronous, - %% the tests need to wait a bit for it to finish. - timer:sleep(100) - end, - send(FUN, null, inet:ntoa(IP)). + {ok, _} = eradius_client_mngr:wanna_send(?SERVER, [one], []), + ok = eradius_client_mngr:reconfigure(?SERVER, #{ip => IP}), + timer:sleep(100), + St = eradius_client_mngr:get_state(?SERVER), + ?equal(#{}, maps:get(pools, St)), + ok. reconf_ports_30(_Config) -> - FUN = fun() -> - eradius_client_mngr:reconfigure(?SERVER, #{no_ports => 30}), - %% socket shutdown is done asynchronous, - %% the tests need to wait a bit for it to finish. - timer:sleep(100) - end, - send(FUN, 30, null). + ok = eradius_client_mngr:reconfigure(?SERVER, #{no_ports => 30}), + St = eradius_client_mngr:get_state(?SERVER), + ?equal(30, maps:get(k_ports, St)), + ok. reconf_ports_10(_Config) -> - FUN = fun() -> - eradius_client_mngr:reconfigure(?SERVER, #{no_ports => 10}), - %% socket shutdown is done asynchronous, - %% the tests need to wait a bit for it to finish. - timer:sleep(100) - end, - send(FUN, 10, null). + ok = eradius_client_mngr:reconfigure(?SERVER, #{no_ports => 10}), + St = eradius_client_mngr:get_state(?SERVER), + ?equal(10, maps:get(k_ports, St)), + ok. send_request_failover(Config) -> Family = proplists:get_value(family, Config), @@ -291,34 +224,22 @@ check_upstream_servers(Config) -> sets:from_list(Servers))), ok. -no_ports_one_wraps() -> - [{doc, "wanna_send must not crash when no_ports = 1 and the req-id wraps past 255"}]. -no_ports_one_wraps(Config) -> - Family = proplists:get_value(family, Config, ipv4), - {ok, _} = application:ensure_all_started(eradius), - Server = #{ip => eradius_test_lib:localhost(Family, native), port => 1812, - secret => <<"secret">>, retries => 3}, - {ok, Client} = - eradius_client_mngr:start_client( - #{family => eradius_test_lib:inet_family(Family), ip => any, no_ports => 1, - servers => #{test_server => Server}}), - %% 257 allocations force the {PortIdx, 255} wrap branch at least once - lists:foreach( - fun(_) -> - ?match({ok, {_Sock, _ReqId, test_server, _Srv, _Info}}, - eradius_client_mngr:wanna_send(Client, [test_server], [])) - end, lists:seq(1, 257)), - ok. - clobber_does_not_hang() -> [{doc, "A pending request whose entry is overwritten by a same-ReqId " "request must still return {error,timeout} to its caller, not hang"}]. clobber_does_not_hang(Config) -> Family = proplists:get_value(family, Config, ipv4), + InetFamily = eradius_test_lib:inet_family(Family), + LH = eradius_test_lib:localhost(Family, native), + BindIP = eradius_test_lib:localhost(Family, mapped), + %% black-hole server: a real UDP socket that never replies, so the connected + %% client socket sees a listener (no ICMP econnrefused) and the clobber is exercised. + {ok, BH} = gen_udp:open(0, [binary, {active, false}, InetFamily, {ip, BindIP}]), + {ok, BHPort} = inet:port(BH), + Peer = {LH, BHPort}, {ok, Sock} = eradius_client_socket:start_link( - #{family => eradius_test_lib:inet_family(Family), active_n => 10}), - %% port 1 on loopback: packets go out, no reply ever comes back - Peer = {eradius_test_lib:localhost(Family, native), 1}, + #{family => InetFamily, active_n => 10, server_addr => Peer, + reqid_reuse_timeout => 30000}), ReqId = 1, Packet = <<1, ReqId, 0, 20, 0:128>>, %% 20-byte minimal RADIUS header Caller = self(), @@ -334,10 +255,260 @@ clobber_does_not_hang(Config) -> eradius_client_socket:send_request(Sock, Peer, ReqId, Packet, 2000) end), %% P1 must not hang; with the bounded call timeout it gets {error,timeout} + Result = + receive + {p1, R} -> R + after 6000 -> + exit(P1, kill), + ct:fail("P1 hung after its pending entry was clobbered") + end, + gen_udp:close(BH), + eradius_client_socket:close(Sock), + ?equal({error, timeout}, Result). + +connected_socket_matches_reply() -> + [{doc, "A connected socket sends without an explicit dest and matches a " + "reply to the pending request by ReqId alone"}]. +connected_socket_matches_reply(Config) -> + Family = proplists:get_value(family, Config, ipv4), + InetFamily = eradius_test_lib:inet_family(Family), + LH = eradius_test_lib:localhost(Family, native), + BindIP = eradius_test_lib:localhost(Family, mapped), + {ok, Server} = gen_udp:open(0, [binary, {active, false}, InetFamily, {ip, BindIP}]), + {ok, SrvPort} = inet:port(Server), + {ok, Sock} = eradius_client_socket:start_link( + #{family => InetFamily, active_n => 10, + server_addr => {LH, SrvPort}, reqid_reuse_timeout => 30000}), + ReqId = 7, + Req = <<1, ReqId, 0, 20, 0:128>>, + Caller = self(), + spawn(fun() -> + R = eradius_client_socket:send_request( + Sock, {LH, SrvPort}, ReqId, Req, 3000), + Caller ! {done, R} + end), + {ok, {FromIP, FromPort, <<_, ReqId, _/binary>>}} = gen_udp:recv(Server, 0, 2000), + Reply = <<2, ReqId, 0, 20, 1:128>>, + ok = gen_udp:send(Server, FromIP, FromPort, Reply), + receive + {done, Result} -> + ?match({ok, <<2, ReqId, _/binary>>, <<>>}, Result) + after 4000 -> + ct:fail("connected socket did not deliver the reply") + end, + gen_udp:close(Server). + +retire_holds_then_closes() -> + [{doc, "A retired socket with no pending requests stays alive during the " + "cooldown and exits normally once it elapses"}]. +retire_holds_then_closes(Config) -> + Family = proplists:get_value(family, Config, ipv4), + InetFamily = eradius_test_lib:inet_family(Family), + LH = eradius_test_lib:localhost(Family, native), + {ok, Sock} = eradius_client_socket:start_link( + #{family => InetFamily, active_n => 10, + server_addr => {LH, 1}, reqid_reuse_timeout => 700}), + MRef = erlang:monitor(process, Sock), + eradius_client_socket:retire(Sock), + timer:sleep(300), + ?equal(true, is_process_alive(Sock)), + receive + {'DOWN', MRef, process, Sock, Reason} -> + ?equal(normal, Reason) + after 2000 -> + ct:fail("retired socket did not close after cooldown") + end. + +retire_waits_for_pending() -> + [{doc, "A retired socket does not close while a request is still pending; " + "it closes after the pending request resolves and cooldown elapsed"}]. +retire_waits_for_pending(Config) -> + Family = proplists:get_value(family, Config, ipv4), + InetFamily = eradius_test_lib:inet_family(Family), + LH = eradius_test_lib:localhost(Family, native), + BindIP = eradius_test_lib:localhost(Family, mapped), + %% black-hole server (real listener, never replies): avoids ICMP econnrefused. + {ok, BH} = gen_udp:open(0, [binary, {active, false}, InetFamily, {ip, BindIP}]), + {ok, BHPort} = inet:port(BH), + Peer = {LH, BHPort}, + %% cooldown 1500ms; force-close at 2x = 3000ms. Retire first so the cooldown + %% starts at t=0, then send a request whose 2500ms timeout resolves between + %% the cooldown (1500ms) and the force-close (3000ms): the close is driven by + %% pending-drain, with comfortable margins around the 1800ms alive-check. + {ok, Sock} = eradius_client_socket:start_link( + #{family => InetFamily, active_n => 10, + server_addr => Peer, reqid_reuse_timeout => 1500}), + MRef = erlang:monitor(process, Sock), + Caller = self(), + eradius_client_socket:retire(Sock), + spawn(fun() -> + R = eradius_client_socket:send_request( + Sock, Peer, 3, <<1, 3, 0, 20, 0:128>>, 2500), + Caller ! {p, R} + end), + %% at ~1800ms the cooldown has elapsed but the request is still pending -> alive + timer:sleep(1800), + ?equal(true, is_process_alive(Sock)), + receive {p, {error, timeout}} -> ok after 4000 -> ct:fail("request never resolved") end, receive - {p1, Result} -> - ?equal({error, timeout}, Result) - after 6000 -> - exit(P1, kill), - ct:fail("P1 hung after its pending entry was clobbered") + {'DOWN', MRef, process, Sock, normal} -> ok + after 4000 -> + ct:fail("retired socket did not close after pending drained") + end, + gen_udp:close(BH). + +client_config_defaults() -> + [{doc, "new client config carries no_ports (K), max_ports and " + "reqid_reuse_timeout with sane defaults"}]. +client_config_defaults(Config) -> + Family = proplists:get_value(family, Config, ipv4), + {ok, _} = application:ensure_all_started(eradius), + Server = #{ip => eradius_test_lib:localhost(Family, native), port => 1812, + secret => <<"secret">>, retries => 3}, + {ok, Client} = + eradius_client_mngr:start_client( + #{family => eradius_test_lib:inet_family(Family), ip => any, + servers => #{test_server => Server}}), + St = eradius_client_mngr:get_state(Client), + ?equal(10, maps:get(k_ports, St)), + ?equal(256, maps:get(max_ports, St)), + ?equal(30000, maps:get(reqid_reuse_timeout, St)), + ok. + +pool_rolls_and_retires() -> + [{doc, "with no_ports=1 the single filler issues ids 0..255 then rolls to a " + "fresh socket on the 257th send; the exhausted socket is retired"}]. +pool_rolls_and_retires(Config) -> + Family = proplists:get_value(family, Config, ipv4), + {ok, _} = application:ensure_all_started(eradius), + Server = #{ip => eradius_test_lib:localhost(Family, native), port => 1812, + secret => <<"secret">>, retries => 3}, + {ok, Client} = + eradius_client_mngr:start_client( + #{family => eradius_test_lib:inet_family(Family), ip => any, + no_ports => 1, reqid_reuse_timeout => 30000, + servers => #{test_server => Server}}), + Allocs = + [begin + {ok, {Pid, ReqId, test_server, _Srv, _Info}} = + eradius_client_mngr:wanna_send(Client, [test_server], []), + {Pid, ReqId} + end || _ <- lists:seq(1, 257)], + {Pids, Ids} = lists:unzip(Allocs), + ?equal(lists:seq(0, 255) ++ [0], Ids), + First256 = lists:sublist(Pids, 256), + ?equal(1, length(lists:usort(First256))), + Socket257 = lists:nth(257, Pids), + ?equal(false, lists:member(Socket257, First256)), + ?equal(2, length(lists:usort(Pids))), + %% the retired socket was told to retire but is still alive (cooling, 30s) + ?equal(true, is_process_alive(hd(First256))), + ok. + +pool_cap_backpressures() -> + [{doc, "with no_ports=1 and max_ports_per_server=2, once both sockets are " + "exhausted-and-cooling wanna_send returns {error, no_ports}"}]. +pool_cap_backpressures(Config) -> + Family = proplists:get_value(family, Config, ipv4), + {ok, _} = application:ensure_all_started(eradius), + Server = #{ip => eradius_test_lib:localhost(Family, native), port => 1812, + secret => <<"secret">>, retries => 3}, + {ok, Client} = + eradius_client_mngr:start_client( + #{family => eradius_test_lib:inet_family(Family), ip => any, + no_ports => 1, max_ports_per_server => 2, + reqid_reuse_timeout => 60000, %% long: cooling sockets stay open + servers => #{test_server => Server}}), + %% 512 allocations exhaust 2 sockets (256 ids each); both go to cooling and + %% cannot be replaced (cap = 2). The 513th allocation must be rejected. + ok = lists:foreach( + fun(_) -> + {ok, {_Pid, _Id, test_server, _S, _I}} = + eradius_client_mngr:wanna_send(Client, [test_server], []) + end, lists:seq(1, 512)), + ?equal({error, no_ports}, + eradius_client_mngr:wanna_send(Client, [test_server], [])), + ok. + +cooling_socket_reclaimed() -> + [{doc, "after a retired socket finishes its cooldown and exits, the manager " + "drops it from the pool (cooling shrinks back to empty)"}]. +cooling_socket_reclaimed(Config) -> + Family = proplists:get_value(family, Config, ipv4), + {ok, _} = application:ensure_all_started(eradius), + Server = #{ip => eradius_test_lib:localhost(Family, native), port => 1812, + secret => <<"secret">>, retries => 3}, + {ok, Client} = + eradius_client_mngr:start_client( + #{family => eradius_test_lib:inet_family(Family), ip => any, + no_ports => 1, reqid_reuse_timeout => 400, + servers => #{test_server => Server}}), + %% 256 allocations exhaust the single filler -> it retires (cooling=1), + %% and a replacement filler opens (active=1). The client has exactly one + %% server, so read its sole pool via maps:values (avoids family-mapped keys). + ok = lists:foreach( + fun(_) -> + {ok, _} = eradius_client_mngr:wanna_send(Client, [test_server], []) + end, lists:seq(1, 256)), + #{pools := P0} = eradius_client_mngr:get_state(Client), + [#{cooling := Cool0}] = maps:values(P0), + ?equal(1, length(Cool0)), + %% wait out the cooldown (400ms); the cooling socket exits and is reclaimed + timer:sleep(1200), + #{pools := P1} = eradius_client_mngr:get_state(Client), + [#{active := Act1, cooling := Cool1}] = maps:values(P1), + ?equal(0, length(Cool1)), + ?equal(1, queue:len(Act1)), + ok. + +rotation_uses_fresh_source_ports() -> + [{doc, "across a 256-id wrap the client sends from a different source port, " + "so no {srcport, ReqId} pair repeats while the old port is cooling"}]. +rotation_uses_fresh_source_ports(Config) -> + Family = proplists:get_value(family, Config, ipv4), + {ok, _} = application:ensure_all_started(eradius), + InetFamily = eradius_test_lib:inet_family(Family), + BindIP = eradius_test_lib:localhost(Family, mapped), + LH = eradius_test_lib:localhost(Family, native), + {ok, Server} = gen_udp:open(0, [binary, {active, false}, InetFamily, {ip, BindIP}]), + {ok, SrvPort} = inet:port(Server), + {ok, Client} = + eradius_client_mngr:start_client( + #{family => InetFamily, ip => any, no_ports => 1, + reqid_reuse_timeout => 60000, + servers => #{s => #{ip => LH, port => SrvPort, + secret => <<"secret">>, retries => 1}}}), + Test = self(), + %% echo server: reply to each request and report the observed source port + _Echo = spawn_link(fun() -> echo_loop(Server, Test) end), + %% 257 synchronous sends: ids run 0..255 on socket 1, then the 257th rolls + %% to a fresh socket (id 0 again) on a new OS-assigned source port. + Pairs = + [begin + {ok, {Pid, ReqId, s, _S, _I}} = + eradius_client_mngr:wanna_send(Client, [s], []), + {ok, _H, _B} = + eradius_client_socket:send_request( + Pid, {LH, SrvPort}, ReqId, <<1, ReqId, 0, 20, 0:128>>, 2000), + receive + {observed, Port, ReqId} -> {Port, ReqId} + after 2000 -> + ct:fail("server did not observe request id ~p", [ReqId]) + end + end || _ <- lists:seq(1, 257)], + gen_udp:close(Server), + %% no {source port, ReqId} pair repeats within the cooldown window + ?equal(length(Pairs), length(lists:usort(Pairs))), + %% the id-0 wrap landed on a second, distinct source port + ?equal(true, length(lists:usort([P || {P, _} <- Pairs])) >= 2), + ok. + +echo_loop(Server, Test) -> + case gen_udp:recv(Server, 0, 5000) of + {ok, {FromIP, FromPort, <<_, ReqId, _/binary>>}} -> + gen_udp:send(Server, FromIP, FromPort, <<2, ReqId, 0, 20, 0:128>>), + Test ! {observed, FromPort, ReqId}, + echo_loop(Server, Test); + {error, _} -> + ok end. diff --git a/test/eradius_metrics_SUITE.erl b/test/eradius_metrics_SUITE.erl index 28455667..9a573872 100644 --- a/test/eradius_metrics_SUITE.erl +++ b/test/eradius_metrics_SUITE.erl @@ -186,11 +186,26 @@ check_single_request(error, EradiusRequestType, _RequestType, _ResponseType) -> ok. check_total_requests(good, N) -> - check_metric(eradius_requests_total, [{server_name, good}], N), - check_metric(eradius_replies_total, [{server_name, good}], N); + check_metric_sum(eradius_requests_total, [{server_name, good}], N), + check_metric_sum(eradius_replies_total, [{server_name, good}], N); check_total_requests(bad, N) -> - check_metric(eradius_requests_total, [{server_name, bad}], N), - check_metric(eradius_replies_total, [{server_name, bad}], N). + check_metric_sum(eradius_requests_total, [{server_name, bad}], N), + check_metric_sum(eradius_replies_total, [{server_name, bad}], N). + +%% Sum a server-side counter across all matching label-sets. The client's +%% per-server pool issues requests from several source ports (round-robin over +%% K fillers), so the server records them under several nas_ip label-sets; the +%% total across those must equal N. +check_metric_sum(Id, Labels, Count) -> + Values = prometheus_counter:values(default, Id), + Filtered = + lists:filter( + fun({ValueLabels, _}) -> Labels -- ValueLabels =:= [] end, + Values), + Sum = lists:sum([V || {_, V} <- Filtered]), + ct:pal("check_metric_sum: ~p, ~p, expect ~p~nFiltered ~p~n", + [Id, Labels, Count, Filtered]), + ?assertEqual(Count, Sum). check_metric_multi({bad_type, accreq}, Id, Labels, _Count) -> Values = prometheus_counter:values(default, Id),