src/lwes_stats.erl

   1 -module (lwes_stats).
   2
   3 % the stats module keeps some stats about what lwes is doing, and provides
   4 % a few ways to roll those stats up.
   5 %
   6 % the key for most methods is one of two forms
   7 %
   8 % - {Label, {Ip, Port}}
   9 % - {Ip, Port}
  10 %
  11 % individual stats are incremented via function calls and the current
  12 % stats can be fetched as a proplist via fetch/1, or as a list of lists
  13 % via rollup/1
  14
  15 %% API
  16 -export([ start_link/0,
  17           initialize/1,            % (Id)
  18           increment_sent/1,        % (Id)
  19           increment_received/1,    % (Id)
  20           increment_errors/1,      % (Id)
  21           % these 2 are only used if validating
  22           increment_considered/1,  % (Id)
  23           increment_validated/1,   % (Id)
  24           delete/1,                % (Id)
  25           fetch/1,                 % (Id)
  26           rollup/1,                % 'id' | 'port' | 'none'
  27           print/1
  28         ]).
  29
  30 %% gen_server callbacks
  31 -export ([init/1,
  32           handle_call/3,
  33           handle_cast/2,
  34           handle_info/2,
  35           terminate/2,
  36           code_change/3]).
  37
  38 % gen_server state
  39 -record (state, {}).
  40 -define (TABLE, lwes_stats).
  41
  42 % stats record for db
  43 -record (stats, {id,       % Id
  44                  sent = 0,
  45                  received = 0,
  46                  errors = 0,
  47                  considered = 0,
  48                  validated = 0
  49                 }).
  50
  51 -define (STATS_SENT_INDEX, #stats.sent).
  52 -define (STATS_RECEIVED_INDEX, #stats.received).
  53 -define (STATS_ERRORS_INDEX, #stats.errors).
  54 -define (STATS_CONSIDERED_INDEX, #stats.considered).
  55 -define (STATS_VALIDATED_INDEX, #stats.validated).
  56
  57 start_link () ->
  58   gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
  59
  60 initialize (Id = {_, {_,_}}) ->
  61   init0 (Id);
  62 initialize (Id = {_, _}) ->
  63   init0 (Id);
  64 initialize (Id) when is_atom(Id) ->
  65   init0 (Id);
  66 initialize (_) ->
  67   erlang:error(badarg).
  68
  69 init0 (Id) ->
  70   ets:insert_new (?TABLE, #stats {id = Id}).
  71
  72 increment_sent (Id) ->
  73   [V] = try_update_counter (Id, ?STATS_SENT_INDEX, 1),
  74   V.
  75
  76 increment_received (Id) ->
  77   [V] = try_update_counter (Id, ?STATS_RECEIVED_INDEX, 1),
  78   V.
  79
  80 increment_errors (Id) ->
  81   [V] = try_update_counter (Id, ?STATS_ERRORS_INDEX, 1),
  82   V.
  83
  84 increment_considered (Id) ->
  85   [V] = try_update_counter (Id, ?STATS_CONSIDERED_INDEX, 1),
  86   V.
  87
  88 increment_validated (Id) ->
  89   [V] = try_update_counter (Id, ?STATS_VALIDATED_INDEX, 1),
  90   V.
  91
  92 update_counter (Id, Index, Amount) ->
  93   ets:update_counter (?TABLE, Id, [{Index,Amount}]).
  94
  95 try_update_counter (Id, Index, Amount) ->
  96   try update_counter (Id, Index, Amount) of
  97     V -> V
  98   catch
  99     error:badarg ->
 100       initialize (Id),
 101       update_counter (Id, Index, Amount)
 102   end.
 103
 104
 105 delete (Id) ->
 106   ets:delete (?TABLE, Id).
 107
 108 fetch (Id) ->
 109   case ets:lookup (?TABLE, Id) of
 110     [] -> undefined;
 111     [#stats { sent = Sent,
 112               received = Received,
 113               errors = Errors,
 114               considered = Considered,
 115               validated = Validated }] ->
 116       [ {sent, Sent},
 117         {received, Received},
 118         {errors, Errors},
 119         {considered, Considered},
 120         {validated, Validated}
 121       ]
 122   end.
 123
 124 pivot_by_key ({Data, Keys}) ->
 125   lists:foldl( fun (Key, A) ->
 126                  [ [ Key, dict:fetch({Key,sent}, Data),
 127                           dict:fetch({Key,received}, Data),
 128                           dict:fetch({Key,errors}, Data),
 129                           dict:fetch({Key,considered}, Data),
 130                           dict:fetch({Key,validated}, Data) ] | A ]
 131                end,
 132                [],
 133                dict:fetch_keys(Keys)).
 134
 135 add_by_key (Key, S, R, E, C, V, {DataIn, KeysIn}) ->
 136   D1 = dict:update_counter({Key,sent},S,DataIn),
 137   D2 = dict:update_counter({Key,received},R,D1),
 138   D3 = dict:update_counter({Key,errors},E,D2),
 139   D4 = dict:update_counter({Key,considered},C,D3),
 140   D5 = dict:update_counter({Key,validated},V,D4),
 141   KeysOut = dict:update_counter(Key,1,KeysIn),
 142   {D5, KeysOut}.
 143
 144 % outputs the table in a two dimensional form which should allow for easy
 145 % construction of output.
 146 % The form is
 147 %
 148 % [ [ id, sent, received, errors, considered, validated],
 149 %   ...
 150 % ]
 151 rollup(label) ->
 152   pivot_by_key(
 153     lists:foldl (
 154       fun (#stats {id = {Label,{_,_}},
 155                    sent = S, received = R, errors = E,
 156                    considered = C, validated = V}, Accum) ->
 157             add_by_key ({Label,{'*','*'}}, S, R, E, C, V, Accum);
 158           (#stats {id = M,
 159                    sent = S, received = R, errors = E,
 160                    considered = C, validated = V}, Accum) when is_atom(M) ->
 161             add_by_key (M, S, R, E, C, V, Accum);
 162           (#stats { sent = S, received = R, errors = E,
 163                     considered = C, validated = V}, Accum) ->
 164             add_by_key ({'_',{'*','*'}}, S, R, E, C, V, Accum)
 165       end,
 166       { dict:new(), dict:new() },
 167       ets:tab2list(?TABLE)
 168     )
 169   );
 170 rollup(port) ->
 171   pivot_by_key(
 172     lists:foldl (
 173       fun (#stats {id = {_,{_,Port}},
 174                    sent = S, received = R, errors = E,
 175                    considered = C, validated = V}, Accum) ->
 176             add_by_key ({'*',{'*',Port}}, S, R, E, C, V, Accum);
 177           (#stats {id = {_, Port},
 178                    sent = S, received = R, errors = E,
 179                    considered = C, validated = V}, Accum) ->
 180             add_by_key ({'*',{'*',Port}}, S, R, E, C, V, Accum);
 181           (#stats {id = M,
 182                    sent = S, received = R, errors = E,
 183                    considered = C, validated = V}, Accum) when is_atom(M) ->
 184             add_by_key (M, S, R, E, C, V, Accum)
 185       end,
 186       { dict:new(), dict:new() },
 187       ets:tab2list(?TABLE)
 188     )
 189   );
 190 rollup(none) ->
 191   [
 192     [ case I of
 193         M when is_atom(M) -> M ;
 194         {_,{_,_}} -> I ;
 195         _ -> {'*',I}
 196       end, S, R, E, C, V ]
 197     || #stats { id = I, sent = S, received = R,
 198                 errors = E, considered = C, validated = V}
 199     <- ets:tab2list(?TABLE)
 200   ].
 201
 202 format_ip_port({'*','*'}) -> io_lib:format("*:~-5s",["*"]);
 203 format_ip_port({'*', Port}) -> io_lib:format("*:~-5b",[Port]);
 204 format_ip_port({Ip, Port}) -> io_lib:format ("~s:~-5b",[lwes_util:ip2bin (Ip), Port]).
 205
 206 print (Type) ->
 207   io:format ("~-21s ~-21s ~10s ~10s ~10s ~10s ~10s~n",
 208              ["label","channel","sent","received",
 209               "errors","considered","validated"]),
 210   io:format ("~-21s ~-21s ~10s ~10s ~10s ~10s ~10s~n",
 211              ["---------------------",
 212               "---------------------",
 213               "----------","----------",
 214               "----------","----------","----------"]),
 215   [
 216     begin
 217       {Label, IpPort} =
 218         case Key of
 219           {L, IP= {_,_}} -> {L, IP};
 220           M when is_atom(M) -> {M, {'*','*'}}
 221         end,
 222       io:format ("~-21w ~-21s ~10b ~10b ~10b ~10b ~10b~n",
 223                [Label, format_ip_port (IpPort),
 224                 Sent, Recv, Err, Cons, Valid]
 225               )
 226     end
 227     || [ Key, Sent, Recv, Err, Cons, Valid ]
 228     <- rollup(Type)
 229   ],
 230   ok.
 231
 232 %%====================================================================
 233 %% gen_server callbacks
 234 %%====================================================================
 235 init([]) ->
 236   ets:new (?TABLE,
 237            [ named_table,
 238              public,
 239              set,
 240              {keypos, 2},
 241              {write_concurrency, true}
 242            ]),
 243   {ok, #state { }}.
 244
 245 handle_call (_Request, _From, State) ->
 246   {reply, ok, State}.
 247
 248 handle_cast (_Request, State) ->
 249   {noreply, State}.
 250
 251 handle_info (_Info, State) ->
 252   {noreply, State}.
 253
 254 terminate (_Reason, _State) ->
 255   ets:delete (?TABLE),
 256   ok.
 257
 258 code_change (_OldVsn, State, _Extra) ->
 259   {ok, State}.
 260
 261 %%--------------------------------------------------------------------
 262 %%% Test functions
 263 %%--------------------------------------------------------------------
 264 -ifdef (TEST).
 265 -include_lib ("eunit/include/eunit.hrl").
 266
 267 setup () ->
 268   case start_link() of
 269     {ok, Pid} -> Pid;
 270     {error, {already_started, _}} -> already_started
 271   end.
 272
 273 cleanup (already_started) -> ok;
 274 cleanup (Pid) -> gen_server:stop (Pid).
 275
 276 fetch_list (Sent, Received, Errors, Considered, Validated) ->
 277   [ {sent, Sent},
 278     {received, Received},
 279     {errors, Errors},
 280     {considered, Considered},
 281     {validated, Validated}
 282   ].
 283
 284 check_entry ([[_,Sent,Recv,Errors,Considered,Validated]],
 285              [[_,Sent,Recv,Errors,Considered,Validated]]) ->
 286   true.
 287
 288 tests_with_id (Id) ->
 289   [
 290     ?_assertEqual(true, initialize(Id)),
 291     % initialization only works once
 292     ?_assertEqual(false, initialize(Id)),
 293     % rollups should show all zeros
 294     ?_assert(check_entry([[dummy,0,0,0,0,0]], rollup(port))),
 295     ?_assert(check_entry([[dummy,0,0,0,0,0]], rollup(label))),
 296     ?_assert(check_entry([[dummy,0,0,0,0,0]], rollup(none))),
 297     % increment all stats and check counts
 298     ?_assertEqual(fetch_list(0,0,0,0,0), fetch(Id)),
 299     ?_assertEqual(1, increment_sent(Id)),
 300     ?_assertEqual(fetch_list(1,0,0,0,0), fetch(Id)),
 301     ?_assertEqual(1, increment_received(Id)),
 302     ?_assertEqual(fetch_list(1,1,0,0,0), fetch(Id)),
 303     ?_assertEqual(2, increment_received(Id)),
 304     ?_assertEqual(fetch_list(1,2,0,0,0), fetch(Id)),
 305     ?_assertEqual(1, increment_errors(Id)),
 306     ?_assertEqual(fetch_list(1,2,1,0,0), fetch(Id)),
 307     ?_assertEqual(1, increment_considered(Id)),
 308     ?_assertEqual(fetch_list(1,2,1,1,0), fetch(Id)),
 309     ?_assertEqual(1, increment_validated(Id)),
 310     ?_assertEqual(fetch_list(1,2,1,1,1), fetch(Id)),
 311     % see that the rollups reflect the same values
 312     ?_assert(check_entry([[dummy,1,2,1,1,1]], rollup(port))),
 313     ?_assert(check_entry([[dummy,1,2,1,1,1]], rollup(label))),
 314     ?_assert(check_entry([[dummy,1,2,1,1,1]], rollup(none))),
 315     % delete the test ids
 316     ?_assertEqual(true, delete(Id)),
 317     ?_assertEqual(true, delete(Id))
 318   ].
 319
 320
 321 lwes_stats_test_ () ->
 322   IdWithLabel = {foo,{{127,0,0,1},9191}},
 323   IdNoLabel = {{127,0,0,1},9191},
 324   BadId = undefined,
 325   NonExistentId = {foo, bar,{{127,0,0,1},9292}},
 326   { setup,
 327     fun setup/0,
 328     fun cleanup/1,
 329     [
 330       % tests with different types of ids
 331       tests_with_id (IdWithLabel),
 332       tests_with_id (IdNoLabel),
 333       % test a few cases with bad ids
 334       ?_assertEqual(undefined, fetch(BadId)),
 335       ?_assertEqual(undefined, fetch(NonExistentId)),
 336       ?_assertException(error, badarg, increment_sent(NonExistentId)),
 337       % these are for additional coverage on the gen_server calls which
 338       % are not currently used for anything
 339       ?_assertEqual({reply, ok, #state{}}, handle_call(foo, bar, #state{})),
 340       ?_assertEqual({noreply, #state{}}, handle_cast(foo, #state{})),
 341       ?_assertEqual({noreply, #state{}}, handle_info(foo, #state{})),
 342       ?_assertEqual({ok, #state{}}, code_change(foo, #state{}, bar)),
 343       % this just tests the case where we might be running inside of a
 344       % larger process, so mostly just here for coverage
 345       ?_assertEqual(ok, cleanup(setup()))
 346     ]
 347   }.
 348
 349 lwes_stats_rollups_test_ () ->
 350   Id1 = {{127,0,0,1},9191},
 351   Id2 = {{127,0,0,1},9192},
 352   Id3 = {foo, Id1},
 353   Id4 = {foo, Id2},
 354   LabelRollupId1 = {foo,{'*','*'}},
 355   LabelRollupId2 = {'_',{'*','*'}},
 356   PortRollupId1 = {'*',{'*',9191}},
 357   PortRollupId2 = {'*',{'*',9192}},
 358   { setup,
 359     fun setup/0,
 360     fun cleanup/1,
 361     { inorder,
 362       [
 363         [ ?_assertEqual(true, initialize(I)) || I <- [ Id1, Id2, Id3, Id4 ] ],
 364         ?_assertEqual(1, increment_sent(Id1)),
 365         ?_assertEqual(1, increment_sent(Id4)),
 366         ?_assertEqual(1, increment_received(Id2)),
 367         ?_assertEqual(1, increment_received(Id3)),
 368         ?_assertEqual(1, increment_errors(Id3)),
 369         ?_assertEqual(1, increment_errors(Id4)),
 370         % check no rollups
 371         ?_assertEqual(lists:sort([[{'*',Id1},1,0,0,0,0],
 372                                   [{'*',Id2},0,1,0,0,0],
 373                                   [Id3,0,1,1,0,0],
 374                                   [Id4,1,0,1,0,0]]),
 375                       lists:sort(rollup(none))),
 376         % check rollup by label
 377         ?_assertEqual(lists:sort([[LabelRollupId1,1,1,2,0,0],
 378                                   [LabelRollupId2,1,1,0,0,0]]),
 379                       lists:sort(rollup(label))),
 380         % check rollup by port
 381         ?_assertEqual(lists:sort([[PortRollupId1,1,1,1,0,0],
 382                                   [PortRollupId2,1,1,1,0,0]]),
 383                       lists:sort(rollup(port)))
 384       ]
 385     }
 386   }.
 387
 388 -endif.