Add latency stats for riak pb client operations [JIRA: RCS-243]

dialyzer.ignore-warnings.ee

@@ -1,7 +1,8 @@
 # Errors
-riak_cs_block_server.erl:312: The pattern Success = {'ok', _} can never match the type {'error',_}
-riak_cs_block_server.erl:347: The pattern {'ok', RiakObject} can never match the type {'error',_}
-riak_cs_pbc.erl:100: The pattern {'ok', ClusterID} can never match the type {'error',_}
+riak_cs_block_server.erl:324: The pattern Success = {'ok', _} can never match the type {'error',_}
+riak_cs_block_server.erl:361: The pattern {'ok', RiakObject} can never match the type {'error',_}
+riak_cs_pbc.erl:58: The variable _ can never match since previous clauses completely covered the type 'pong'
+riak_cs_pbc.erl:175: The pattern {'ok', ClusterID} can never match the type {'error',_}
 # Warnings
 Unknown functions:
   app_helper:get_prop_or_env/3

include/riak_cs_stats.hrl

@@ -0,0 +1,29 @@
+%% ---------------------------------------------------------------------
+%%
+%% Copyright (c) 2015 Basho Technologies, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you under the Apache License,
+%% Version 2.0 (the "License"); you may not use this file
+%% except in compliance with the License.  You may obtain
+%% a copy of the License at
+%%
+%%   http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% ---------------------------------------------------------------------
+
+-define(WITH_STATS(StatsKey, Statement),
+        begin
+            _ = riak_cs_stats:inflow(StatsKey),
+            StartTime__with_stats = os:timestamp(),
+            Result__with_stats = Statement,
+            _ = riak_cs_stats:update_with_start(StatsKey, StartTime__with_stats,
+                                                Result__with_stats),
+            Result__with_stats
+        end).

priv/tools/repair_gc_bucket.erl

@@ -166,7 +166,7 @@ process_gc_keys(Pbc, Options, Continuation, [GCKey | Keys]) ->
                                          {error, term()}.
 repair_manifests_for_gc_key(Pbc, Options, GCKey) ->
     Timeout = riak_cs_config:get_gckey_timeout(),
-    case riak_cs_pbc:get_object(Pbc, ?GC_BUCKET, GCKey, Timeout) of
+    case riak_cs_pbc:get_no_stats(Pbc, ?GC_BUCKET, GCKey, [], Timeout) of
         {ok, GCObj} ->
             FileSet = riak_cs_gc:decode_and_merge_siblings(
                         GCObj, twop_set:new()),

riak_test/tests/stats_test.erl

@@ -46,10 +46,10 @@ confirm() ->
     lager:info("Confirming stats"),
     Stats1 = query_stats(UserConfig, rtcs:cs_port(hd(RiakNodes))),
     _Stats2 = status_cmd(),
-    confirm_stat_count(Stats1, <<"service_get_buckets">>, 2),
-    confirm_stat_count(Stats1, <<"object_gets">>, 1),
-    confirm_stat_count(Stats1, <<"object_puts">>, 1),
-    confirm_stat_count(Stats1, <<"object_deletes">>, 1),
+    confirm_stat_count(Stats1, <<"service_get_out_one">>, 2),
+    confirm_stat_count(Stats1, <<"object_get_out_one">>, 1),
+    confirm_stat_count(Stats1, <<"object_put_out_one">>, 1),
+    confirm_stat_count(Stats1, <<"object_delete_out_one">>, 1),
     rtcs:pass().
 
 status_cmd() ->
@@ -72,35 +72,32 @@ query_stats(UserConfig, Port) ->
 
 confirm_initial_stats(StatData) ->
     %% Check for values for all meters to be 0 when system is initially started
-    ?assertEqual(614, length(StatData)),
-    [?assert(proplists:is_defined(StatType, StatData))
-     || StatType <- [<<"block_gets">>,
-                     <<"block_puts">>,
-                     <<"block_deletes">>,
-                     <<"service_get_buckets">>,
-                     <<"bucket_list_keys">>,
-                     <<"bucket_creates">>,
-                     <<"bucket_deletes">>,
-                     <<"bucket_get_acl">>,
-                     <<"bucket_put_acl">>,
-                     <<"object_gets">>,
-                     <<"object_puts">>,
-                     <<"object_heads">>,
-                     <<"object_deletes">>,
-                     <<"object_get_acl">>,
-                     <<"object_put_acl">>]],
-
-    Exceptions = [<<"request_pool_workers">>,
-                  <<"request_pool_size">>,
-                  <<"bucket_list_pool_workers">>],
-    ShouldBeZeros = lists:foldl(fun proplists:delete/2, StatData, Exceptions),
+    ?assertEqual(1096, length(StatData)),
     [begin
-         lager:debug("testing ~p:~p", [Name, Value]),
+         StatKey = list_to_binary(StatType ++ "_out_one"),
+         lager:debug("StatKey: ~p~n", [StatKey]),
+         ?assert(proplists:is_defined(StatKey, StatData)),
+         Value = proplists:get_value(StatKey, StatData),
          ?assertEqual(0, Value)
-     end|| {Name, Value} <- ShouldBeZeros],
+     end || StatType <- ["service_get",
+                         "list_objects_get",
+                         "bucket_put",
+                         "bucket_delete",
+                         "bucket_acl_get",
+                         "bucket_acl_put",
+                         "object_get",
+                         "object_put",
+                         "object_head",
+                         "object_delete",
+                         "object_acl_get",
+                         "object_acl_put",
+                         "riakc_get_block_n_one",
+                         "riakc_put_block",
+                         "riakc_delete_block_constrained"
+                        ]],
 
     lager:debug("~p", [proplists:get_value(<<"request_pool_workers">>, StatData)]),
-    ?assertEqual(rtcs:request_pool_size()-1,
+    ?assertEqual(rtcs:request_pool_size() - 1,
                  proplists:get_value(<<"request_pool_workers">>, StatData)),
     ?assertEqual(rtcs:bucket_list_pool_size(),
                  proplists:get_value(<<"bucket_list_pool_workers">>, StatData)).

src/riak_cs_access.erl

@@ -136,15 +136,16 @@ merge_stats(Stats, Acc) ->
 %% which the keys are Riak CS node names.  The value for each key is a
 %% list of samples.  Each sample is an orddict full of metrics.
 -spec get_usage(riak_client(),
-                term(), %% TODO: riak_cs:user_key() type doesn't exist
+                term(),     %% TODO: riak_cs:user_key() type doesn't exist
                 boolean(),  %% Not used in this module
                 calendar:datetime(),
                 calendar:datetime()) ->
                        {Usage::orddict:orddict(), Errors::[{slice(), term()}]}.
 get_usage(RcPid, User, _AdminAccess, Start, End) ->
     {ok, Period} = archive_period(),
-    {Usage, Errors} = rts:find_samples(RcPid, ?ACCESS_BUCKET, User,
-                                       Start, End, Period),
+    RtsPuller = riak_cs_riak_client:rts_puller(
+                  RcPid, ?ACCESS_BUCKET, User, [riakc, get_access]),
+    {Usage, Errors} = rts:find_samples(RtsPuller, Start, End, Period),
     {group_by_node(Usage), Errors}.
 
 group_by_node(Samples) ->

src/riak_cs_access_archiver.erl

@@ -202,7 +202,8 @@ archive_user(User, RcPid, Table, Slice) ->
 
 store(User, RcPid, Record, Slice) ->
     {ok, MasterPbc} = riak_cs_riak_client:master_pbc(RcPid),
-    case catch riakc_pb_socket:put(MasterPbc, Record) of
+    Timeout = riak_cs_config:put_access_timeout(),
+    case catch riak_cs_pbc:put(MasterPbc, Record, Timeout, [riakc, put_access]) of
         ok ->
             ok = lager:debug("Archived access stats for ~s ~p",
                              [User, Slice]);

src/riak_cs_acl.erl

@@ -310,8 +310,8 @@ owner_id(?ACL{owner=Owner}, _) ->
 owner_id(#acl_v1{owner=OwnerData}, RcPid) ->
     {Name, CanonicalId} = OwnerData,
     case riak_cs_user:get_user_by_index(?ID_INDEX,
-                                           list_to_binary(CanonicalId),
-                                           RcPid) of
+                                        list_to_binary(CanonicalId),
+                                        RcPid) of
         {ok, {Owner, _}} ->
             Owner?RCS_USER.key_id;
         {error, _} ->