Metrics for :random strategy (#11)

* add metrics for random

* fix dialyze warning

* fix test

* add missing listener stop

* remove tet changes

* bump gun version

* remove unused helper

* report size

* add try catch

Author: wwitek-whatnot

Makefile

@@ -2,7 +2,7 @@
 
 PROJECT = whatnot_gun
 PROJECT_DESCRIPTION = HTTP/1.1, HTTP/2 and Websocket client for Erlang/OTP.
-PROJECT_VERSION = 2.4.3
+PROJECT_VERSION = 2.4.4
 
 # The OTP application is renamed to whatnot_gun for the Whatnot fork, but the
 # modules, application callback, and supervisor keep their upstream `gun_*`

ebin/whatnot_gun.app

@@ -1,6 +1,6 @@
 {application, 'whatnot_gun', [
 	{description, "HTTP/1.1, HTTP/2 and Websocket client for Erlang/OTP."},
-	{vsn, "2.4.3"},
+	{vsn, "2.4.4"},
 	{modules, ['gun','gun_app','gun_conns_sup','gun_content_handler','gun_cookies','gun_cookies_list','gun_data_h','gun_default_event_h','gun_event','gun_http','gun_http2','gun_http3','gun_pool','gun_pool_events_h','gun_pools_sup','gun_protocols','gun_public_suffix','gun_quicer','gun_raw','gun_socks','gun_sse_h','gun_sup','gun_tcp','gun_tcp_proxy','gun_tls','gun_tls_proxy','gun_tls_proxy_cb','gun_tls_proxy_http2_connect','gun_tunnel','gun_ws','gun_ws_h','gun_ws_protocol']},
 	{registered, [gun_sup]},
 	{applications, [kernel,stdlib,public_key,ssl,cowlib]},

src/gun_app.erl

@@ -24,6 +24,9 @@
 
 start(_Type, _Args) ->
 	gun_pools = ets:new(gun_pools, [ordered_set, public, named_table]),
+	gun_pool_counters = ets:new(gun_pool_counters, [
+		set, public, named_table, {read_concurrency, true}
+	]),
 	gun_sup:start_link().
 
 stop(_State) ->

src/gun_pool.erl

@@ -23,6 +23,8 @@
 -export([info/0]).
 -export([info/1]).
 -export([info/2]).
+-export([metrics/0]).
+-export([metrics/2]).
 -export([await_up/1]).
 -export([await_up/2]).
 -export([checkout/2]). %% Use responsibly!
@@ -151,11 +153,12 @@
 	conns :: #{pid() => down | {setup, any()} | {up, http | http2 | ws | raw, map()}},
 	conns_meta = #{} :: meta(),
 	await_up = [] :: [{pid(), any()}],
-	lookup :: #{strategy := random, table := ets:tid()}
+	lookup :: #{strategy := random, table := ets:tid(),
+	            active := atomics:atomics_ref(), pool_key := {any(), binary()}}
 	        | #{strategy := least_loaded, available := gb_trees:tree(),
 	            stream_counts := #{pid() => non_neg_integer()},
 	            seqs := #{pid() => non_neg_integer()}, next_seq := non_neg_integer(),
-	            up_count := non_neg_integer()}
+	            up_count := non_neg_integer(), manager_pid := pid()}
 }).
 
 %% Pool management.
@@ -210,6 +213,55 @@ info(Authority, Scope) ->
 			gen_statem:call(ManagerPid, info)
 	end.
 
+%% Read pool metrics from ETS without calling the manager (random pools only).
+%% Counters may reflect a torn snapshot across connections but no call is made.
+-spec metrics() -> [map()].
+metrics() ->
+	[M || {PoolKey, Row} <- ets:tab2list(gun_pool_counters),
+		M <- [metrics_from_row(PoolKey, Row)], M =/= undefined].
+
+-spec metrics(binary(), any()) -> undefined | map().
+metrics(Authority, Scope) ->
+	metrics_by_key({Scope, Authority}).
+
+metrics_by_key(PoolKey) ->
+	case ets:lookup(gun_pool_counters, PoolKey) of
+		[] ->
+			undefined;
+		[{_, Row}] ->
+			metrics_from_row(PoolKey, Row)
+	end.
+
+metrics_from_row({Scope, _}, #{table := Tid, max_streams := Max, active := Active, size := Size}) ->
+	%% The per-pool table is owned by the manager and deleted on terminate.
+	try
+		ActiveN = atomics:get(Active, 1),
+		StreamRows = ets:tab2list(Tid),
+		Streams = lists:sum([C || {_, C} <- StreamRows]),
+		{Full, High} = occupancy(StreamRows, Max),
+		#{
+			scope => Scope,
+			size => Size,
+			active => ActiveN,
+			max_streams => Max,
+			streams => Streams,
+			full => Full,
+			high => High
+		}
+	catch _:_ ->
+		undefined
+	end.
+
+occupancy(_Rows, infinity) ->
+	{0, 0};
+occupancy(Rows, Max) ->
+	HighThreshold = (Max * 9 + 9) div 10,
+	lists:foldl(fun({_, Count}, {F, H}) ->
+		F1 = if Count >= Max -> F + 1; true -> F end,
+		H1 = if Count >= HighThreshold -> H + 1; true -> H end,
+		{F1, H1}
+	end, {0, 0}, Rows).
+
 -spec await_up(pid() | binary()) -> ok | {error, pool_not_found, atom()}.
 await_up(ManagerPid) when is_pid(ManagerPid) ->
 	gen_statem:call(ManagerPid, await_up, 5000);
@@ -526,13 +578,21 @@ start_link(Host, Port, Opts) ->
 
 init({Host, Port, Opts}) ->
 	process_flag(trap_exit, true),
-	true = ets:insert_new(gun_pools, {gun_pools_key(Host, Port, Opts), self()}),
+	PoolKey = gun_pools_key(Host, Port, Opts),
+	true = ets:insert_new(gun_pools, {PoolKey, self()}),
 	Size = maps:get(size, Opts, 8),
 	%% @todo Only start processes in static mode.
 	Lookup = case maps:get(lookup_strategy, Opts, random) of
 		random ->
 			Tid = ets:new(gun_pooled_conns, [ordered_set, public]),
-			#{strategy => random, table => Tid};
+			Active = atomics:new(1, [{signed, false}]),
+			ets:insert(gun_pool_counters, {PoolKey, #{
+				table => Tid,
+				size => Size,
+				max_streams => infinity,
+				active => Active
+			}}),
+			#{strategy => random, table => Tid, active => Active, pool_key => PoolKey};
 		least_loaded ->
 		#{strategy => least_loaded, available => gb_trees:empty(),
 			stream_counts => #{}, seqs => #{}, next_seq => 0,
@@ -630,6 +690,7 @@ degraded_setup(ConnPid, Msg, StateData0=#state{conns=Conns, conns_meta=ConnsMeta
 		%% Websocket or tunnel stream refs.
 		{up, Protocol, Meta} ->
 			Settings = #{},
+			metrics_conn_up(Lookup, Protocol),
 			StateData = StateData0#state{
 				conns=Conns#{ConnPid => {up, Protocol, Settings}},
 				conns_meta=ConnsMeta#{ConnPid => Meta},
@@ -681,13 +742,15 @@ handle_common({call, From}, {checkout, _ReqOpts}, _,
 	end;
 handle_common(cast, {release_stream, ConnPid}, _, StateData) ->
 	{keep_state, adjust_stream_count(ConnPid, -1, StateData)};
-handle_common(info, {gun_notify, ConnPid, settings_changed, Settings}, _, StateData=#state{conns=Conns}) ->
+handle_common(info, {gun_notify, ConnPid, settings_changed, Settings}, _, StateData=#state{conns=Conns, lookup=Lookup}) ->
 	%% Assert that the state is correct.
 	{up, http2, _} = maps:get(ConnPid, Conns),
+	metrics_update_max_streams(Lookup, Settings),
 	{keep_state, StateData#state{conns=Conns#{ConnPid => {up, http2, Settings}}}};
 handle_common(info, {gun_down, ConnPid, Protocol, _Reason, _KilledStreams}, _,
-		StateData=#state{lookup=#{strategy := random}, conns=Conns}) ->
+		StateData=#state{lookup=#{strategy := random, active := Active}, conns=Conns}) ->
 	{up, Protocol, _} = maps:get(ConnPid, Conns),
+	atomics:sub(Active, 1, 1),
 	{next_state, degraded, StateData#state{conns=Conns#{ConnPid => down}}};
 handle_common(info, {gun_down, ConnPid, Protocol, _Reason, _KilledStreams}, _,
 		StateData=#state{lookup=#{strategy := least_loaded, available := Available,
@@ -706,6 +769,15 @@ handle_common(info, {gun_down, ConnPid, Protocol, _Reason, _KilledStreams}, _,
 	}};
 %% @todo We do not want to reconnect automatically when the pool is dynamic.
 handle_common(info, {'DOWN', _MRef, process, ConnPid0, Reason}, _, StateData0) ->
+	%% gun_down fires before DOWN in normal cases and already decrements active.
+	%% Guard against sudden process death (no gun_down) where the conn is still up.
+	_ = case {maps:get(ConnPid0, StateData0#state.conns, undefined),
+			StateData0#state.lookup} of
+		{{up, _, _}, #{strategy := random, active := Active}} ->
+			atomics:sub(Active, 1, 1);
+		_ ->
+			ok
+	end,
 	StateData=#state{host=Host, port=Port, opts=Opts, lookup=Lookup, conns=Conns}
 		= remove_down_conn(ConnPid0, StateData0),
 	case Reason of
@@ -873,5 +945,29 @@ terminate(Reason, StateName, #state{host=Host, port=Port, opts=Opts, await_up=Aw
 	gen_statem:reply([
 		{reply, ReplyTo, {error, {terminate, StateName, Reason}}}
 	|| ReplyTo <- AwaitUp]),
-	true = ets:delete(gun_pools, gun_pools_key(Host, Port, Opts)),
+	PoolKey = gun_pools_key(Host, Port, Opts),
+	true = ets:delete(gun_pools, PoolKey),
+	ets:delete(gun_pool_counters, PoolKey),
 	ok.
+
+metrics_conn_up(#{strategy := random, active := Active, pool_key := PoolKey}, http) ->
+	atomics:add(Active, 1, 1),
+	update_pool_counter_max_streams(PoolKey, 1);
+metrics_conn_up(#{strategy := random, active := Active}, _Protocol) ->
+	atomics:add(Active, 1, 1);
+metrics_conn_up(_Lookup, _Protocol) ->
+	ok.
+
+metrics_update_max_streams(#{strategy := random, pool_key := PoolKey}, Settings) ->
+	MaxStreams = maps:get(max_concurrent_streams, Settings, infinity),
+	update_pool_counter_max_streams(PoolKey, MaxStreams);
+metrics_update_max_streams(_Lookup, _Settings) ->
+	ok.
+
+update_pool_counter_max_streams(PoolKey, MaxStreams) ->
+	case ets:lookup(gun_pool_counters, PoolKey) of
+		[{_, Row}] ->
+			ets:insert(gun_pool_counters, {PoolKey, Row#{max_streams => MaxStreams}});
+		[] ->
+			ok
+	end.

test/pool_SUITE.erl

@@ -41,8 +41,8 @@ groups() ->
 		stop_pool,
 		degraded_configuration_error
 	],
-	[{random, [], Tests},
-	 {least_loaded, [], Tests ++ [least_loaded_routing, least_loaded_round_robin]}].
+	[{random, [], Tests ++ [metrics_idle, metrics_streams]},
+	 {least_loaded, [], Tests ++ [least_loaded_routing, least_loaded_round_robin, metrics_undefined]}].
 
 init_per_suite(Config) ->
 	{ok, _} = cowboy:start_clear({?MODULE, tcp}, [], do_proto_opts()),
@@ -57,11 +57,13 @@ end_per_suite(_) ->
 		max_streams_h2_size_2_random,
 		max_streams_h2_size_2_retry_random,
 		reconnect_h1_random,
+		metrics_streams_random,
 		max_streams_h2_size_1_least_loaded,
 		max_streams_h2_size_1_retry_least_loaded,
 		max_streams_h2_size_2_least_loaded,
 		max_streams_h2_size_2_retry_least_loaded,
-		reconnect_h1_least_loaded
+		reconnect_h1_least_loaded,
+		least_loaded_routing_least_loaded
 	],
 	_ = [cowboy:stop_listener(Listener) || Listener <- ExtraListeners],
 	ok.
@@ -543,6 +545,84 @@ least_loaded_round_robin(Config) ->
 	%% Every connection must have been used exactly once.
 	Size = length(lists:usort(ConnPids)).
 
+metrics_idle(Config) ->
+	doc("Confirm pool metrics reflect active connections and zero streams when idle."),
+	Port = config(port, Config),
+	Authority = iolist_to_binary(["localhost:", integer_to_binary(Port)]),
+	Scope = scope(?FUNCTION_NAME, Config),
+	Size = 3,
+	{ok, ManagerPid} = gun_pool:start_pool("localhost", Port, #{
+		conn_opts => #{protocols => [http2]},
+		scope => Scope,
+		lookup_strategy => random,
+		size => Size
+	}),
+	gun_pool:await_up(ManagerPid),
+	%% Make a request so SETTINGS have been negotiated.
+	{async, PoolStreamRef} = gun_pool:get("/",
+		#{<<"host">> => Authority}, #{scope => Scope}),
+	{response, nofin, 200, _} = gun_pool:await(PoolStreamRef),
+	{ok, _} = gun_pool:await_body(PoolStreamRef),
+	%% Flush pending manager messages (e.g. settings_changed) with a synchronous call.
+	gun_pool:info(Authority, Scope),
+	#{size := Size, active := Size, streams := 0, full := 0, high := 0} = gun_pool:metrics(Authority, Scope).
+
+metrics_streams(Config) ->
+	doc("Confirm pool metrics count in-flight streams and detect full/high connections."),
+	Listener = listener_name(?FUNCTION_NAME, Config),
+	ProtoOpts = do_proto_opts(),
+	{ok, _} = cowboy:start_clear(Listener, [], ProtoOpts#{max_concurrent_streams => 5}),
+	Port = ranch:get_port(Listener),
+	Authority = iolist_to_binary(["localhost:", integer_to_binary(Port)]),
+	Scope = scope(?FUNCTION_NAME, Config),
+	{ok, ManagerPid} = gun_pool:start_pool("localhost", Port, #{
+		conn_opts => #{protocols => [http2]},
+		scope => Scope,
+		lookup_strategy => random,
+		size => 1
+	}),
+	gun_pool:await_up(ManagerPid),
+	%% Fill the single connection to max capacity.
+	[gun_pool:get("/delay", #{<<"host">> => Authority},
+		#{scope => Scope}) || _ <- lists:seq(1, 5)],
+	%% Poll until event handler has counted all 5 streams and manager has processed SETTINGS.
+	ok = wait_for_metrics(Authority, Scope, #{size => 1, active => 1, streams => 5, full => 1,
+		high => 1, max_streams => 5}).
+
+metrics_undefined(Config) ->
+	doc("Confirm metrics returns undefined for the least_loaded strategy."),
+	Port = config(port, Config),
+	Authority = iolist_to_binary(["localhost:", integer_to_binary(Port)]),
+	Scope = scope(?FUNCTION_NAME, Config),
+	{ok, ManagerPid} = gun_pool:start_pool("localhost", Port, #{
+		conn_opts => #{protocols => [http2]},
+		scope => Scope,
+		lookup_strategy => least_loaded,
+		size => 1
+	}),
+	gun_pool:await_up(ManagerPid),
+	undefined = gun_pool:metrics(Authority, Scope).
+
+%% Poll until metrics match the expected map.
+%% Calls info/2 each iteration to flush pending manager messages (e.g. settings_changed).
+wait_for_metrics(Authority, Scope, Expected) ->
+	wait_for_metrics(Authority, Scope, Expected, 50).
+
+wait_for_metrics(_Authority, _Scope, _Expected, 0) ->
+	{error, timeout};
+wait_for_metrics(Authority, Scope, Expected, N) ->
+	gun_pool:info(Authority, Scope),
+	Actual = gun_pool:metrics(Authority, Scope),
+	Match = is_map(Actual) andalso maps:fold(
+		fun(K, V, Acc) -> Acc andalso maps:get(K, Actual, undefined) =:= V end,
+		true, Expected),
+	case Match of
+		true -> ok;
+		false ->
+			timer:sleep(20),
+			wait_for_metrics(Authority, Scope, Expected, N - 1)
+	end.
+
 %% Poll the manager until every connection's stream count is back to 0.
 wait_all_streams_released(ManagerPid) ->
 	wait_all_streams_released(ManagerPid, 100).