Merge pull request #1194 from basho/feature/stats-sys-mochi-pools

Add status around PB pools, memory, system and mochiweb [RCS-244]

Reviewed-by: kuenishi

riak_test/tests/access_stats_test.erl

@@ -100,7 +100,7 @@ assert_access_stats(Format, UserConfig, {Begin, End}) ->
     ?assertEqual(  1, sum_samples(Format, "BucketRead",   "Count", Samples)),
     ?assertEqual(  1, sum_samples(Format, "KeyDelete",    "Count", Samples)),
     ?assertEqual(  1, sum_samples(Format, "BucketDelete", "Count", Samples)),
-    pass.
+    rtcs:pass().
 
 verify_stats_lost_logging(UserConfig, RiakNodes, CSNodes) ->
     KeyId = UserConfig#aws_config.access_key_id,
@@ -115,7 +115,7 @@ verify_stats_lost_logging(UserConfig, RiakNodes, CSNodes) ->
     ExpectLine = io_lib:format("lost access stat: User=~s, Slice=", [KeyId]),
     lager:debug("expected log line: ~s", [ExpectLine]),
     true = rt:expect_in_log(CSNode, ExpectLine),
-    pass.
+    rtcs:pass().
 
 
 node_samples_from_content(json, Node, Content) ->

riak_test/tests/auth_bypass_test.erl

@@ -30,21 +30,33 @@ confirm() ->
     KeyId = UserConfig#aws_config.access_key_id,
     Port = rtcs:cs_port(hd(RiakNodes)),
 
-    confirm_auth_bypass("riak-cs", "stats", UserConfig, Port),
+    confirm_auth_bypass_for_stats("riak-cs", "stats", UserConfig, Port),
     confirm_auth_bypass("riak-cs", "users", UserConfig, Port),
     confirm_auth_bypass("riak-cs", "user/"  ++ KeyId, UserConfig, Port),
-    confirm_auth_bypass("riak-cs", "usage/" ++ KeyId ++ "/ab/" ++ rtcs:datetime() ++ "/" ++ rtcs:datetime(),
+    confirm_auth_bypass("riak-cs", "usage/" ++ KeyId ++ "/ab/" ++
+                            rtcs:datetime() ++ "/" ++ rtcs:datetime(),
                         UserConfig, Port),
-    pass.
+    rtcs:pass().
+
+confirm_auth_bypass_for_stats(Bucket, Key, UserConfig, Port) ->
+    {S3Content, CurlContent} = get_both_contents(Bucket, Key, UserConfig, Port),
+    S3Json = drop_volatile_stats_keys(mochijson2:decode(S3Content)),
+    CurlJson = drop_volatile_stats_keys(mochijson2:decode(CurlContent)),
+    ?assertEqual([], S3Json -- CurlJson),
+    ?assertEqual([], CurlJson -- S3Json).
 
 confirm_auth_bypass(Bucket, Key, UserConfig, Port) ->
+    {S3Content, CurlContent} = get_both_contents(Bucket, Key, UserConfig, Port),
+    ?assertEqual(S3Content, CurlContent).
+
+get_both_contents(Bucket, Key, UserConfig, Port) ->
     S3Result = erlcloud_s3:get_object(Bucket, Key, UserConfig),
-    S3Content = proplists:get_value(content, S3Result),
+    S3Content = extract_contents(proplists:get_value(content, S3Result)),
     lager:debug("erlcloud output: ~p~n", [S3Content]),
 
-    CurlContent = curl_request(Bucket, Key, Port),
+    CurlContent = extract_contents(curl_request(Bucket, Key, Port)),
     lager:debug("curl output: ~p~n", [CurlContent]),
-    ?assertEqual(extract_contents(S3Content), extract_contents(CurlContent)).
+    {S3Content, CurlContent}.
 
 curl_request(Bucket, Key, Port) ->
     Cmd = "curl -s http://localhost:" ++ integer_to_list(Port)
@@ -71,3 +83,21 @@ extract_contents([Boundary | Tokens], Boundary, Contents) ->
     extract_contents(Tokens, Boundary, Contents);
 extract_contents([_ | Tokens], Boundary, Contents) ->
     extract_contents(Tokens, Boundary, Contents).
+
+drop_volatile_stats_keys({struct, KVs}) ->
+    [{K, V} || {K, V} <- KVs, not lists:member(K, volatile_stats_keys())].
+
+volatile_stats_keys() ->
+    [<<"pbc_pool_master_workers">>,
+     <<"pbc_pool_master_size">>,
+     <<"object_web_active_sockets">>,
+     <<"memory_total">>,
+     <<"memory_processes">>,
+     <<"memory_processes_used">>,
+     <<"memory_system">>,
+     <<"memory_atom_used">>,
+     <<"memory_binary">>,
+     <<"memory_ets">>,
+     <<"sys_monitor_count">>,
+     <<"sys_port_count">>,
+     <<"sys_process_count">>].

riak_test/tests/block_audit_test.erl

@@ -40,7 +40,7 @@ confirm() ->
         {multibag, _} ->
             lager:info("Block audit script does not supprt multibag env."),
             lager:info("Skip the test."),
-            pass;
+            rtcs:pass();
         _ -> confirm1()
     end.
 
@@ -74,7 +74,7 @@ confirm1() ->
     BlockKeysFileList = [filename:join([Home, "actual-orphaned-blocks", B]) ||
                         B <- [?BUCKET1, ?BUCKET2]],
     tools_helper:offline_delete({RiakNodes, CSNodes, Stanchion}, BlockKeysFileList),
-    pass.
+    rtcs:pass().
 
 setup_objects(RiakNodes, UserConfig, Bucket, Type,
               KeyAlive, KeyOrphaned, KeyFalseOrphaned) ->

riak_test/tests/buckets_test.erl

@@ -61,7 +61,7 @@ confirm() ->
 
     ok = verify_max_buckets_per_user(UserConfig),
 
-    pass.
+    rtcs:pass().
 
 
 verify_create_delete(UserConfig) ->

riak_test/tests/gc_tests.erl

@@ -83,7 +83,7 @@ confirm() ->
     setup_normal_obj([{"spam", 42}, {"ham", 65536}, {"egg", 7}], UserConfig),
 
     verify_partial_gc_run(hd(CSNodes), RiakNodes, Start, End),
-    pass.
+    rtcs:pass().
 
 setup_normal_obj(ObjSpecs, UserConfig) ->
     %% Put and delete some objects

riak_test/tests/legacy_s3_rewrite_test.erl

@@ -46,7 +46,7 @@ confirm() ->
     WaitTime = 2 * rt_config:get(rt_max_wait_time),
     case rtcs:cmd(Cmd, [{cd, CsSrcDir}, {env, Env}, {args, Args}], WaitTime) of
         ok ->
-            pass;
+            rtcs:pass();
         {error, Reason} ->
             lager:error("Error : ~p", [Reason]),
             error({?MODULE, Reason})

riak_test/tests/list_objects_v2_test.erl

@@ -32,7 +32,7 @@ confirm() ->
     pass = list_objects_test_helper:test(UserConfig),
 
     ok = list_to_non_existent_bucket_many_times(RiakNodes),
-    pass.
+    rtcs:pass().
 
 assert_v2_is_default(CSNodes) ->
     true = rpc:call(hd(CSNodes), riak_cs_list_objects_utils, fold_objects_for_list_keys, []),

riak_test/tests/migration_15_to_20_test.erl

@@ -54,19 +54,19 @@ confirm(upgrade_with_full_ops) ->
     {ok, InitialState} = cs_suites:new(SetupRes),
     {ok, EvolvedState} = cs_suites:fold_with_state(InitialState, upgrade_history()),
     {ok, _FinalState}  = cs_suites:cleanup(EvolvedState),
-    pass;
+    rtcs:pass();
 confirm(upgrade_with_reduced_ops) ->
     SetupRes = setup_previous(),
     {ok, InitialState} = cs_suites:new(SetupRes, rtcs:reduced_ops()),
     {ok, EvolvedState} = cs_suites:fold_with_state(InitialState, upgrade_history()),
     {ok, _FinalState}  = cs_suites:cleanup(EvolvedState),
-    pass;
+    rtcs:pass();
 confirm(no_upgrade_with_reduced_ops) ->
     SetupRes = rtcs:setup(2, rtcs:configs(custom_configs(current))),
     {ok, InitialState} = cs_suites:new(SetupRes, cs_suites:reduced_ops()),
     {ok, EvolvedState} = cs_suites:fold_with_state(InitialState, no_upgrade_history()),
     {ok, _FinalState}  = cs_suites:cleanup(EvolvedState),
-    pass.
+    rtcs:pass().
 
 setup_previous() ->
     PrevConfigs = rtcs:previous_configs(custom_configs(previous)),

riak_test/tests/migration_mixed_test.erl

@@ -41,7 +41,7 @@ confirm() ->
     {ok, InitialState} = cs_suites:new(SetupRes),
     {ok, EvolvedState} = cs_suites:fold_with_state(InitialState, history()),
     {ok, _FinalState}  = cs_suites:cleanup(EvolvedState),
-    pass.
+    rtcs:pass().
 
 setup_previous() ->
     PrevConfigs = rtcs:previous_configs(custom_configs(previous)),

riak_test/tests/select_gc_bucket_test.erl

@@ -39,7 +39,7 @@ confirm() ->
         {multibag, _} ->
             lager:info("select_gc_bucket script does not supprt multibag env."),
             lager:info("Skip the test."),
-            pass;
+            rtcs:pass();
         _ -> confirm1()
     end.
 
@@ -67,7 +67,7 @@ confirm1() ->
     lager:debug("select_gc_bucket.erl log:============= END"),
 
     tools_helper:offline_delete({RiakNodes, CSNodes, Stanchion}, [BlockKeysFile]),
-    pass.
+    rtcs:pass().
 
 upload_object(UserConfig, Bucket, normal, Key) ->
     SingleBlock = crypto:rand_bytes(400),

riak_test/tests/sibling_benchmark.erl

@@ -114,7 +114,7 @@ confirm() ->
     ?assertEqual(ok, erlcloud_s3:delete_bucket(?TEST_BUCKET, UserConfig)),
     lager:info("User is valid on the cluster, and has no buckets"),
     ?assertEqual([{buckets, []}], erlcloud_s3:list_buckets(UserConfig)),
-    pass.
+    rtcs:pass().
 
 start_object_reader(UserConfig) ->
     Pid = spawn_link(fun() -> object_reader(UserConfig, 1) end),

riak_test/tests/stats_test.erl

@@ -71,8 +71,8 @@ query_stats(UserConfig, Port) ->
     JsonData.
 
 confirm_initial_stats(StatData) ->
-    %% Check for values for all meters to be 0 when system is initially started
-    ?assertEqual(1096, length(StatData)),
+    lager:debug("length(StatData) = ~p", [length(StatData)]),
+    ?assert(1125 < length(StatData)),
     [begin
          StatKey = list_to_binary(StatType ++ "_out_one"),
          lager:debug("StatKey: ~p~n", [StatKey]),

src/riak_cs_riak_client.erl

@@ -25,7 +25,8 @@
 %% API
 -export([checkout/0, checkout/1,
          checkin/1, checkin/2]).
--export([pbc_pool_name/1,
+-export([pbc_pools/0,
+         pbc_pool_name/1,
          rts_puller/4]).
 -export([
          stop/1,
@@ -106,6 +107,10 @@ checkin(Pool, RcPid) ->
     ok = gen_server:call(RcPid, cleanup),
     poolboy:checkin(Pool, RcPid).
 
+-spec pbc_pools() -> [atom()].
+pbc_pools() ->
+    [pbc_pool_name(B) || {B, _, _} <- riak_cs_mb_helper:bags()].
+
 -spec pbc_pool_name(master | bag_id()) -> atom().
 pbc_pool_name(master) ->
     pbc_pool_master;

src/riak_cs_stats.erl

@@ -34,7 +34,14 @@
 
 %% Lower level API, mainly for debugging or investigation from shell
 -export([report_exometer_item/3,
-         report_pool/1]).
+         report_pool/0,
+         report_pool/1,
+         report_mochiweb/0,
+         report_memory/0,
+         report_system/0,
+         system_monitor_count/0,
+         system_version/0,
+         system_architecture/0]).
 
 -export([init/0]).
 
@@ -191,8 +198,9 @@ get_stats() ->
         [report_exometer_item(Key, SubKey, ExometerType) ||
             Key <- counting_metrics(),
             {SubKey, ExometerType} <- counting_subkeys()],
-    PoolStats = [report_pool(P) || P <- [request_pool, bucket_list_pool]],
-    lists:flatten([DurationStats, CountingStats, PoolStats]).
+    lists:flatten([DurationStats, CountingStats,
+                   report_pool(), report_mochiweb(),
+                   report_memory(), report_system()]).
 
 %% ====================================================================
 %% Internal
@@ -239,21 +247,81 @@ datapoints(spiral) ->
     [one, count].
 
 suffixes(histogram) ->
-    ["_mean", "_median", "_95", "_99", "_100"];
+    ["mean", "median", "95", "99", "100"];
 suffixes(spiral) ->
-    ["_one", "_total"].
+    ["one", "total"].
+
+-spec report_pool() -> [[{atom(), integer()}]].
+report_pool() ->
+    Pools = [request_pool, bucket_list_pool | riak_cs_riak_client:pbc_pools()],
+    [report_pool(Pool) || Pool <- Pools].
 
 -spec report_pool(atom()) -> [{atom(), integer()}].
 report_pool(Pool) ->
     {_PoolState, PoolWorkers, PoolOverflow, PoolSize} = poolboy:status(Pool),
-    Name = binary_to_list(atom_to_binary(Pool, latin1)),
-    [{list_to_atom(lists:flatten([Name, $_, "workers"])), PoolWorkers},
-     {list_to_atom(lists:flatten([Name, $_, "overflow"])), PoolOverflow},
-     {list_to_atom(lists:flatten([Name, $_, "size"])), PoolSize}].
+    [{metric_to_atom([Pool], "workers"), PoolWorkers},
+     {metric_to_atom([Pool], "overflow"), PoolOverflow},
+     {metric_to_atom([Pool], "size"), PoolSize}].
+
+-spec report_mochiweb() -> [[{atom(), integer()}]].
+report_mochiweb() ->
+    MochiIds = [object_web, admin_web],
+    [report_mochiweb(Id) || Id <- MochiIds].
+
+report_mochiweb(Id) ->
+    Children = supervisor:which_children(riak_cs_sup),
+    case lists:keyfind(Id, 1, Children) of
+        false -> [];
+        {_, Pid, _, _} -> report_mochiweb(Id, Pid)
+    end.
 
+report_mochiweb(Id, Pid) ->
+    [{metric_to_atom([Id], PropKey), gen_server:call(Pid, {get, PropKey})} ||
+        PropKey <- [active_sockets, waiting_acceptors, port]].
+
+-spec report_memory() -> [{atom(), integer()}].
+report_memory() ->
+    lists:map(fun({K, V}) -> {metric_to_atom([memory], K), V} end, erlang:memory()).
+
+-spec report_system() -> [{atom(), integer()}].
+report_system() ->
+    [{nodename, erlang:node()},
+     {connected_nodes, erlang:nodes()},
+     {sys_driver_version, list_to_binary(erlang:system_info(driver_version))},
+     {sys_heap_type, erlang:system_info(heap_type)},
+     {sys_logical_processors, erlang:system_info(logical_processors)},
+     {sys_monitor_count, system_monitor_count()},
+     {sys_otp_release, list_to_binary(erlang:system_info(otp_release))},
+     {sys_port_count, erlang:system_info(port_count)},
+     {sys_process_count, erlang:system_info(process_count)},
+     {sys_smp_support, erlang:system_info(smp_support)},
+     {sys_system_version, system_version()},
+     {sys_system_architecture, system_architecture()},
+     {sys_threads_enabled, erlang:system_info(threads)},
+     {sys_thread_pool_size, erlang:system_info(thread_pool_size)},
+     {sys_wordsize, erlang:system_info(wordsize)}].
+
+system_monitor_count() ->
+    lists:foldl(fun(Pid, Count) ->
+                        case erlang:process_info(Pid, monitors) of
+                            {monitors, Mons} ->
+                                Count + length(Mons);
+                            _ ->
+                                Count
+                        end
+                end, 0, processes()).
+
+system_version() ->
+    list_to_binary(string:strip(erlang:system_info(system_version), right, $\n)).
+
+system_architecture() ->
+    list_to_binary(erlang:system_info(system_architecture)).
+
+metric_to_atom(Key, Suffix) when is_atom(Suffix) ->
+    metric_to_atom(Key, atom_to_list(Suffix));
 metric_to_atom(Key, Suffix) ->
     StringKey = string:join([atom_to_list(Token) || Token <- Key], "_"),
-    list_to_atom(lists:flatten([StringKey, Suffix])).
+    list_to_atom(lists:flatten([StringKey, $_, Suffix])).
 
 -ifdef(TEST).