beam_stats_consumer_statsd_SUITE.erl

   1 -module(beam_stats_consumer_statsd_SUITE).
   2
   3 -include_lib("beam_stats/include/beam_stats.hrl").
   4 -include_lib("beam_stats/include/beam_stats_ets_table.hrl").
   5 -include_lib("beam_stats/include/beam_stats_process.hrl").
   6 -include_lib("beam_stats/include/beam_stats_process_ancestry.hrl").
   7 -include_lib("beam_stats/include/beam_stats_processes.hrl").
   8
   9 -export(
  10     [ all/0
  11     , groups/0
  12     ]).
  13
  14 %% Test cases
  15 -export(
  16     [ t_full_cycle/1
  17     ]).
  18
  19 -define(GROUP, beam_stats_consumer_statsd).
  20
  21 %% ============================================================================
  22 %% Common Test callbacks
  23 %% ============================================================================
  24
  25 all() ->
  26     [{group, ?GROUP}].
  27
  28 groups() ->
  29     Tests =
  30         [ t_full_cycle
  31         ],
  32     Properties = [],
  33     [{?GROUP, Properties, Tests}].
  34
  35 %% ============================================================================
  36 %%  Test cases
  37 %% ============================================================================
  38
  39 t_full_cycle(_Cfg) ->
  40     meck:new(beam_stats_source),
  41     BEAMStatsExpected = meck_expect_beam_stats(),
  42     BEAMStatsComputed = beam_stats_state:export(beam_stats_state:new()),
  43     ct:log("BEAMStatsExpected: ~p~n", [BEAMStatsExpected]),
  44     ct:log("BEAMStatsComputed: ~p~n", [BEAMStatsComputed]),
  45     BEAMStatsExpected = BEAMStatsComputed,
  46
  47     {ok,[hope,beam_stats]} = application:ensure_all_started(beam_stats),
  48     ct:log("beam_stats started~n"),
  49     ServerPort = 8125,
  50     {ok, ServerSocket} = gen_udp:open(ServerPort, [binary, {active, false}]),
  51     ct:log("UDP server started started~n"),
  52     {ok, _} = beam_stats_consumer:add(beam_stats_consumer_statsd,
  53         [ {consumption_interval , 60000}
  54         , {dst_host             , "localhost"}
  55         , {dst_port             , ServerPort}
  56         , {src_port             , 8124}
  57         , {num_msgs_per_packet  , 10}
  58         ]
  59     ),
  60     ct:log("consumer added~n"),
  61     _ = meck_expect_beam_stats(
  62         % Double the original values, so that deltas will equal originals after
  63         % 1 update of new beam_stats_state:t()
  64         [ {io_bytes_in      , 6}
  65         , {io_bytes_out     , 14}
  66         , {context_switches , 10}
  67         ]
  68     ),
  69     ct:log("meck_expect_beam_stats ok~n"),
  70     {} = beam_stats_producer:sync_produce_consume(),
  71     ct:log("produced and consumed~n"),
  72     ok = application:stop(beam_stats),
  73     ct:log("beam_stats stopped~n"),
  74
  75     ResultOfReceive1 = gen_udp:recv(ServerSocket, 0),
  76     ResultOfReceive2 = gen_udp:recv(ServerSocket, 0),
  77     ResultOfReceive3 = gen_udp:recv(ServerSocket, 0),
  78     ResultOfReceive4 = gen_udp:recv(ServerSocket, 0),
  79     ok = gen_udp:close(ServerSocket),
  80     {ok, {_, _, PacketReceived1}} = ResultOfReceive1,
  81     {ok, {_, _, PacketReceived2}} = ResultOfReceive2,
  82     {ok, {_, _, PacketReceived3}} = ResultOfReceive3,
  83     {ok, {_, _, PacketReceived4}} = ResultOfReceive4,
  84     ct:log("PacketReceived1: ~n~s~n", [PacketReceived1]),
  85     ct:log("PacketReceived2: ~n~s~n", [PacketReceived2]),
  86     ct:log("PacketReceived3: ~n~s~n", [PacketReceived3]),
  87     ct:log("PacketReceived4: ~n~s~n", [PacketReceived4]),
  88     PacketsCombined =
  89         << PacketReceived1/binary
  90          , PacketReceived2/binary
  91          , PacketReceived3/binary
  92          , PacketReceived4/binary
  93         >>,
  94     ct:log("PacketsCombined: ~n~s~n", [PacketsCombined]),
  95     MsgsExpected =
  96         [ <<"beam_stats.node_foo_host_bar.io.bytes_in:3|g">>
  97         , <<"beam_stats.node_foo_host_bar.io.bytes_out:7|g">>
  98         , <<"beam_stats.node_foo_host_bar.context_switches:5|g">>
  99         , <<"beam_stats.node_foo_host_bar.reductions:9|g">>
 100         , <<"beam_stats.node_foo_host_bar.run_queue:17|g">>
 101         , <<"beam_stats.node_foo_host_bar.memory.mem_type_foo:1|g">>
 102         , <<"beam_stats.node_foo_host_bar.memory.mem_type_bar:2|g">>
 103         , <<"beam_stats.node_foo_host_bar.memory.mem_type_baz:3|g">>
 104         , <<"beam_stats.node_foo_host_bar.ets_table.size.foo.foo:5|g">>
 105         , <<"beam_stats.node_foo_host_bar.ets_table.memory.foo.foo:40|g">>
 106         , <<"beam_stats.node_foo_host_bar.ets_table.size.bar.37:8|g">>
 107         , <<"beam_stats.node_foo_host_bar.ets_table.memory.bar.37:64|g">>
 108
 109         % Processes totals
 110         , <<"beam_stats.node_foo_host_bar.processes_count_all:3|g">>
 111         , <<"beam_stats.node_foo_host_bar.processes_count_exiting:0|g">>
 112         , <<"beam_stats.node_foo_host_bar.processes_count_garbage_collecting:0|g">>
 113         , <<"beam_stats.node_foo_host_bar.processes_count_registered:1|g">>
 114         , <<"beam_stats.node_foo_host_bar.processes_count_runnable:0|g">>
 115         , <<"beam_stats.node_foo_host_bar.processes_count_running:3|g">>
 116         , <<"beam_stats.node_foo_host_bar.processes_count_suspended:0|g">>
 117         , <<"beam_stats.node_foo_host_bar.processes_count_waiting:0|g">>
 118
 119         % Process 1
 120         , <<"beam_stats.node_foo_host_bar.process_memory.reg_name_foo.0_1_0:15|g">>
 121         , <<"beam_stats.node_foo_host_bar.process_total_heap_size.reg_name_foo.0_1_0:25|g">>
 122         , <<"beam_stats.node_foo_host_bar.process_stack_size.reg_name_foo.0_1_0:10|g">>
 123         , <<"beam_stats.node_foo_host_bar.process_message_queue_len.reg_name_foo.0_1_0:0|g">>
 124
 125         % Process 2
 126         , <<"beam_stats.node_foo_host_bar.process_memory.bar_mod-bar_fun-1--NONE--NONE.0_2_0:25|g">>
 127         , <<"beam_stats.node_foo_host_bar.process_total_heap_size.bar_mod-bar_fun-1--NONE--NONE.0_2_0:35|g">>
 128         , <<"beam_stats.node_foo_host_bar.process_stack_size.bar_mod-bar_fun-1--NONE--NONE.0_2_0:40|g">>
 129         , <<"beam_stats.node_foo_host_bar.process_message_queue_len.bar_mod-bar_fun-1--NONE--NONE.0_2_0:5|g">>
 130
 131         % Process 3
 132         , <<"beam_stats.node_foo_host_bar.process_memory.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:25|g">>
 133         , <<"beam_stats.node_foo_host_bar.process_total_heap_size.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:35|g">>
 134         , <<"beam_stats.node_foo_host_bar.process_stack_size.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:40|g">>
 135         , <<"beam_stats.node_foo_host_bar.process_message_queue_len.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:1|g">>
 136         ],
 137     MsgsReceived = binary:split(PacketsCombined, <<"\n">>, [global, trim]),
 138     RemoveExpectedFromReceived =
 139         fun (Expected, Received) ->
 140             ct:log(
 141                 "Looking for expected msg ~p in remaining received ~p~n",
 142                 [Expected, Received]
 143             ),
 144             true = lists:member(Expected, Received),
 145             Received -- [Expected]
 146         end,
 147     [] = lists:foldl(RemoveExpectedFromReceived, MsgsReceived, MsgsExpected),
 148     meck:unload(beam_stats_source).
 149
 150 meck_expect_beam_stats() ->
 151     meck_expect_beam_stats([]).
 152
 153 meck_expect_beam_stats(Overrides) ->
 154     IOBytesIn       = hope_kv_list:get(Overrides, io_bytes_in , 3),
 155     IOBytesOut      = hope_kv_list:get(Overrides, io_bytes_out, 7),
 156     ContextSwitches = hope_kv_list:get(Overrides, context_switches, 5),
 157     Pid0 = list_to_pid("<0.0.0>"),
 158     Pid1 = list_to_pid("<0.1.0>"),
 159     Pid2 = list_to_pid("<0.2.0>"),
 160     Pid3 = list_to_pid("<0.3.0>"),
 161     Process1 =
 162         #beam_stats_process
 163         { pid               = Pid1
 164         , registered_name   = {some, reg_name_foo}
 165         , ancestry =
 166               #beam_stats_process_ancestry
 167               { raw_initial_call  = {foo_mod, foo_fun, 2}
 168               , otp_initial_call  = none
 169               , otp_ancestors     = none
 170               }
 171         , status            = running
 172         , memory            = 15
 173         , total_heap_size   = 25
 174         , stack_size        = 10
 175         , message_queue_len = 0
 176         },
 177     Process2 =
 178         #beam_stats_process
 179         { pid               = Pid2
 180         , registered_name   = none
 181         , ancestry =
 182               #beam_stats_process_ancestry
 183               { raw_initial_call  = {bar_mod, bar_fun, 1}
 184               , otp_initial_call  = none
 185               , otp_ancestors     = none
 186               }
 187         , status            = running
 188         , memory            = 25
 189         , total_heap_size   = 35
 190         , stack_size        = 40
 191         , message_queue_len = 5
 192         },
 193     Process3 =
 194         #beam_stats_process
 195         { pid               = Pid3
 196         , registered_name   = none
 197         , ancestry =
 198               #beam_stats_process_ancestry
 199               { raw_initial_call  = {baz_mod, baz_fun, 3}
 200               , otp_initial_call  = {some, {baz_otp_mod, baz_otp_fun, 2}}
 201               , otp_ancestors     = {some, [Pid0, Pid1]}
 202               }
 203         , status            = running
 204         , memory            = 25
 205         , total_heap_size   = 35
 206         , stack_size        = 40
 207         , message_queue_len = 1
 208         },
 209     Processes =
 210         #beam_stats_processes
 211         { individual_stats =
 212             [ Process1
 213             , Process2
 214             , Process3
 215             ]
 216         , count_all                = 3
 217         , count_exiting            = 0
 218         , count_garbage_collecting = 0
 219         , count_registered         = 1
 220         , count_runnable           = 0
 221         , count_running            = 3
 222         , count_suspended          = 0
 223         , count_waiting            = 0
 224         },
 225     ETSTableStatsFoo =
 226         #beam_stats_ets_table
 227         { id     = foo
 228         , name   = foo
 229         , size   = 5
 230         , memory = 40
 231         },
 232     ETSTableStatsBar =
 233         #beam_stats_ets_table
 234         { id     = 37
 235         , name   = bar
 236         , size   = 8
 237         , memory = 64
 238         },
 239     meck:expect(beam_stats_source, erlang_memory,
 240         fun () -> [{mem_type_foo, 1}, {mem_type_bar, 2}, {mem_type_baz, 3}] end),
 241     meck:expect(beam_stats_source, erlang_node,
 242         fun () -> 'node_foo@host_bar' end),
 243     meck:expect(beam_stats_source, erlang_registered,
 244         fun () -> [reg_name_foo] end),
 245     meck:expect(beam_stats_source, erlang_statistics,
 246         fun (io              ) -> {{input, IOBytesIn}, {output, IOBytesOut}}
 247         ;   (context_switches) -> {ContextSwitches, 0}
 248         ;   (reductions      ) -> {0, 9} % 1st element is unused
 249         ;   (run_queue       ) -> 17
 250         end
 251     ),
 252     meck:expect(beam_stats_source, ets_all,
 253         fun () -> [foo, 37] end),
 254     meck:expect(beam_stats_source, erlang_system_info,
 255         fun (wordsize) -> 8 end),
 256     meck:expect(beam_stats_source, ets_info,
 257         fun (foo, memory) -> 5
 258         ;   (foo, name  ) -> foo
 259         ;   (foo, size  ) -> 5
 260         ;   (37 , memory) -> 8
 261         ;   (37 , name  ) -> bar
 262         ;   (37 , size  ) -> 8
 263         end
 264     ),
 265     meck:expect(beam_stats_source, erlang_processes,
 266         fun () -> [Pid1, Pid2, Pid3] end),
 267     meck:expect(beam_stats_source, os_timestamp,
 268         fun () -> {1, 2, 3} end),
 269     meck:expect(beam_stats_source, erlang_process_info,
 270         fun (P, K) when P == Pid1 ->
 271                 case K
 272                 of  dictionary        -> {K, []}
 273                 ;   initial_call      -> {K, {foo_mod, foo_fun, 2}}
 274                 ;   registered_name   -> {K, reg_name_foo}
 275                 ;   status            -> {K, running}
 276                 ;   memory            -> {K, 15}
 277                 ;   total_heap_size   -> {K, 25}
 278                 ;   stack_size        -> {K, 10}
 279                 ;   message_queue_len -> {K, 0}
 280                 end
 281         ;   (P, K) when P == Pid2 ->
 282                 case K
 283                 of  dictionary        -> {K, []}
 284                 ;   initial_call      -> {K, {bar_mod, bar_fun, 1}}
 285                 ;   registered_name   -> []
 286                 ;   status            -> {K, running}
 287                 ;   memory            -> {K, 25}
 288                 ;   total_heap_size   -> {K, 35}
 289                 ;   stack_size        -> {K, 40}
 290                 ;   message_queue_len -> {K, 5}
 291                 end
 292         ;   (P, K) when P == Pid3 ->
 293                 Dict =
 294                     [ {'$initial_call', {baz_otp_mod, baz_otp_fun, 2}}
 295                     , {'$ancestors'   , [Pid0, Pid1]}
 296                     ],
 297                 case K
 298                 of  dictionary        -> {K, Dict}
 299                 ;   initial_call      -> {K, {baz_mod, baz_fun, 3}}
 300                 ;   registered_name   -> []
 301                 ;   status            -> {K, running}
 302                 ;   memory            -> {K, 25}
 303                 ;   total_heap_size   -> {K, 35}
 304                 ;   stack_size        -> {K, 40}
 305                 ;   message_queue_len -> {K, 1}
 306                 end
 307         end
 308     ),
 309     #beam_stats
 310     { timestamp = {1, 2, 3}
 311     , node_id   = 'node_foo@host_bar'
 312     , memory    = [{mem_type_foo, 1}, {mem_type_bar, 2}, {mem_type_baz, 3}]
 313     , io_bytes_in  = IOBytesIn
 314     , io_bytes_out = IOBytesOut
 315     , context_switches = ContextSwitches
 316     , reductions       = 9
 317     , run_queue        = 17
 318     , ets              = [ETSTableStatsFoo, ETSTableStatsBar]
 319     , processes        = Processes
 320     }.