Fix cull_members, more robustness against bad pool names, test metrics

src/pooler.erl

@@ -142,7 +142,7 @@ pool_stats() ->
 %% EXPERIMENTAL
 %%
 -spec cull_pool(string(), non_neg_integer()) -> ok.
-cull_pool(PoolName, MaxAgeMin) when MaxAgeMin > 0 ->
+cull_pool(PoolName, MaxAgeMin) when MaxAgeMin >= 0 ->
     gen_server:call(?SERVER, {cull_pool, PoolName, MaxAgeMin}).
 
 %% ------------------------------------------------------------------
@@ -301,15 +301,18 @@ add_pids(PoolName, N, State) ->
 -spec take_member(string(), pid(), #state{}) ->
     {error_no_pool | error_no_members | pid(), #state{}}.
 take_member(PoolName, From, #state{pools = Pools} = State) ->
-    send_metric(pool_metric(PoolName, take_rate), 1, meter),
     take_member_from_pool(fetch_pool(PoolName, Pools), From, State).
 
 take_member_from_pool(error_no_pool, _From, State) ->
     {error_no_pool, State};
-take_member_from_pool(#pool{name = PoolName} = Pool, From,
+take_member_from_pool(#pool{name = PoolName,
+                            max_count = Max,
+                            free_pids = Free,
+                            in_use_count = NumInUse,
+                            free_count = NumFree} = Pool,
+                      From,
                       #state{pools = Pools, consumer_to_pid = CPMap} = State) ->
-    #pool{max_count = Max, free_pids = Free, in_use_count = NumInUse,
-          free_count = NumFree} = Pool,
+    send_metric(pool_metric(PoolName, take_rate), 1, meter),
     case Free of
         [] when NumInUse =:= Max ->
             send_metric(<<"pooler.error_no_members_count">>, {inc, 1}, counter),
@@ -505,16 +508,17 @@ cull_members(PoolName, MaxAgeMin, #state{pools = Pools} = State) ->
     cull_members_from_pool(fetch_pool(PoolName, Pools), MaxAgeMin, State).
 
 -spec cull_members_from_pool(#pool{}, non_neg_integer(), #state{}) -> #state{}.
-cull_members_from_pool(#pool{} = Pool, MaxAgeMin,
+cull_members_from_pool(#pool{free_count = FreeCount,
+                             init_count = InitCount,
+                             in_use_count = InUseCount} = Pool, MaxAgeMin,
                        #state{all_members = AllMembers} = State) ->
-    MaxCull = Pool#pool.free_count - Pool#pool.init_count,
+    MaxCull = FreeCount - (InitCount - InUseCount),
     case MaxCull > 0 of
         true ->
             MemberInfo = member_info(Pool#pool.free_pids, AllMembers),
             ExpiredMembers =
                 expired_free_members(MemberInfo, os:timestamp(), MaxAgeMin),
             CullList = lists:sublist(ExpiredMembers, MaxCull),
-            % FIXME: need the member pid, not just pid info
             lists:foldl(fun({CullMe, _}, S) -> remove_pid(CullMe, S) end,
                         State, CullList);
         false ->
@@ -530,7 +534,7 @@ member_info(Pids, AllMembers) ->
 expired_free_members(Members, Now, MaxAgeMin) ->
     Micros = 60 * 1000 * 1000,
     [ MI || MI = {_, {_, free, LastReturn}} <- Members,
-            timer:now_diff(Now, LastReturn) > (MaxAgeMin * Micros) ].
+            timer:now_diff(Now, LastReturn) >= (MaxAgeMin * Micros) ].
 
 
 -spec send_metric(binary(), term(), atom()) -> ok.
@@ -539,7 +543,7 @@ expired_free_members(Members, Now, MaxAgeMin) ->
 send_metric(Name, Value, Type) ->
     case application:get_env(pooler, metrics_module) of
         undefined -> ok;
-        Mod -> Mod:notify(Name, Value, Type)
+        {ok, Mod} -> Mod:notify(Name, Value, Type)
     end,
     ok.
 

test/fake_metrics.erl

@@ -0,0 +1,76 @@
+-module(fake_metrics).
+-behaviour(gen_server).
+-define(SERVER, ?MODULE).
+
+-include_lib("eunit/include/eunit.hrl").
+%% ------------------------------------------------------------------
+%% API Function Exports
+%% ------------------------------------------------------------------
+
+-export([start_link/0,
+         notify/3,
+         get_metrics/0,
+         reset_metrics/0,
+         stop/0
+        ]).
+
+%% ------------------------------------------------------------------
+%% gen_server Function Exports
+%% ------------------------------------------------------------------
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+%% ------------------------------------------------------------------
+%% API Function Definitions
+%% ------------------------------------------------------------------
+
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+notify(Name, Value, Type) ->
+    gen_server:cast(?SERVER, {Name, Value, Type}).
+
+reset_metrics() ->
+    gen_server:call(?SERVER, reset).
+
+stop() ->
+    gen_server:call(?SERVER, stop).
+
+get_metrics() ->
+    gen_server:call(?SERVER, get_metrics).
+
+%% ------------------------------------------------------------------
+%% gen_server Function Definitions
+%% ------------------------------------------------------------------
+-record(state, {
+          metrics = []
+         }).
+
+init(_) ->
+    {ok, #state{}}.
+
+handle_call(reset, _From, State) ->
+    {reply, ok, State#state{metrics = []}};
+handle_call(get_metrics, _From, #state{metrics = Metrics}=State) ->
+    {reply, Metrics, State};
+handle_call(stop, _From, State) ->
+    {stop, normal, stop_ok, State};
+handle_call(_Request, _From, State) ->
+    erlang:error({what, _Request}),
+    {noreply, ok, State}.
+
+handle_cast({_N, _V, _T}=M, #state{metrics = Metrics} = State) ->
+    {noreply, State#state{metrics = [M|Metrics]}};
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+

test/pooler_test.erl

@@ -102,6 +102,14 @@ assert_tc_valid(Pid) ->
 %     stop_tc(Pid1).
 
 pooler_basics_test_() ->
+    {setup,
+     fun() ->
+             application:set_env(pooler, metrics_module, fake_metrics),
+             fake_metrics:start_link()
+     end,
+     fun(_X) ->
+             fake_metrics:stop()
+     end,
     {foreach,
      % setup
      fun() ->
@@ -232,9 +240,49 @@ pooler_basics_test_() ->
        fun() ->
                ?assertEqual(ok, pooler:cull_pool("p1", 10))
        end
-      }
+      },
+
+      {"cull_pool culls unused members",
+       fun() ->
+               %% take all
+               [P1, P2, _P3] = [pooler:take_member(), pooler:take_member(), pooler:take_member()],
+               %% return one
+               pooler:return_member(P1),
+               pooler:return_member(P2),
+               %% call a sync action since return_member is async
+               _Ignore = pooler:pool_stats(),
+               ?assertEqual(ok, pooler:cull_pool("p1", 0)),
+               ?assertEqual(2, length(pooler:pool_stats()))
+       end
+      },
 
-     ]}.
+      {"metrics have been called",
+       fun() ->
+               %% exercise the API to ensure we have certain keys reported as metrics
+               fake_metrics:reset_metrics(),
+               Pids = [ pooler:take_member() || _I <- lists:seq(1, 10) ],
+               [ pooler:return_member(P) || P <- Pids ],
+               pooler:take_member("bad_pool_name"),
+               %% kill and unused member
+               exit(hd(Pids), kill),
+               %% kill a used member
+               KillMe = pooler:take_member("p1"),
+               exit(KillMe, kill),
+               %% FIXME: We need to wait for pooler to process the
+               %% exit message. This is ugly, will fix later.
+               timer:sleep(200),                % :(
+               ExpectKeys = [<<"pooler.error_no_members_count">>,
+                             <<"pooler.events">>,
+                             <<"pooler.killed_free_count">>,
+                             <<"pooler.killed_in_use_count">>,
+                             <<"pooler.p1.free_count">>,
+                             <<"pooler.p1.in_use_count">>,
+                             <<"pooler.p1.take_rate">>],
+               Metrics = fake_metrics:get_metrics(),
+               GotKeys = lists:usort([ Name || {Name, _, _} <- Metrics ]),
+               ?assertEqual(ExpectKeys, GotKeys)
+       end}
+     ]}}.
 
 
 pooler_integration_test_() ->