anchor.diff:                                           Acc
                                        end
                   end, [], Strs).
  +
  +%% This function changes the group leader around a call to error_logger. The
  +%% reason this code is needed is that when these functions are called via
  +%% rpc:call, the group leader is on the node making the rpc call, in the case
  +%% of riak-admin, this is nodetool running as an escript. Since nodetool has
  +%% error logging configured, this makes sure that the log message ends up in
  +%% riaks's log like we want.
  +log_to_riak_node(Format, Args) ->
  +    GL = erlang:group_leader(),
  +    erlang:group_leader(whereis(user), self()),
  +    error_logger:error_msg(Format, Args),
  +    erlang:group_leader(GL, self()).
  
positive.code:     remove_node(list_to_atom(Node)).
  
  remove_node(Node) when is_atom(Node) ->
      {ok, C} = riak:local_client(),
      Res = C:remove_from_cluster(Node),
      io:format("~p\n", [Res]).
  
  -spec(status([]) -> ok).
  status([]) ->
      case riak_kv_status:statistics() of
          [] ->
              io:format("riak_kv_stat is not enabled.\n", []);
          Stats ->
              StatString = format_stats(Stats,
                                        ["-------------------------------------------\n",
                                         io_lib:format("1-minute stats for ~p~n",[node()])]),
              io:format("~s\n", [StatString])
      end.
  
  reip([OldNode, NewNode]) ->
      %% reip is called when node is down (so riak_core_ring_manager is not running),
      %% so it has to use the basic ring operations.
      %%
      %% Do *not* convert to use riak_core_ring_manager:ring_trans.
      %%
      application:load(riak_core),
      RingStateDir = app_helper:get_env(riak_core, ring_state_dir),
      {ok, RingFile} = riak_core_ring_manager:find_latest_ringfile(),
      BackupFN = filename:join([RingStateDir, filename:basename(RingFile)++".BAK"]),
      {ok, _} = file:copy(RingFile, BackupFN),
      io:format("Backed up existing ring file to ~p~n", [BackupFN]),
      Ring = riak_core_ring_manager:read_ringfile(RingFile),
      NewRing = riak_core_ring:rename_node(Ring, OldNode, NewNode),
      riak_core_ring_manager:do_write_ringfile(NewRing),
      io:format("New ring file written to ~p~n", 
                [element(2, riak_core_ring_manager:find_latest_ringfile())]).
  
  %% Check if all nodes in the cluster agree on the partition assignment
  -spec(ringready([]) -> ok | error).
  ringready([]) ->
      case riak_kv_status:ringready() of
          {ok, Nodes} ->
              io:format("TRUE All nodes agree on the ring ~p\n", [Nodes]);
          {error, {different_owners, N1, N2}} ->
              io:format("FALSE Node ~p and ~p list different partition owners\n", [N1, N2]),
              error;
          {error, {nodes_down, Down}} ->
              io:format("FALSE ~p down.  All nodes need to be up to check.\n", [Down]),
              error
      end.
  
  
negative.code: %% and list any owned vnodes that are *not* running
  -spec(transfers([]) -> ok).
  transfers([]) ->
      try
          {DownNodes, Pending} = riak_kv_status:transfers(),
          case DownNodes of
              [] -> ok;
              _  -> io:format("Nodes ~p are currently down.\n", [DownNodes])
          end,
          F = fun({waiting_to_handoff, Node, Count}, Acc) ->
                  io:format("~p waiting to handoff ~p partitions\n", [Node, Count]),
                  Acc + 1;
              ({stopped, Node, Count}, Acc) ->
                  io:format("~p does not have ~p primary partitions running\n", [Node, Count]),
                  Acc + 1
          end,
          case lists:foldl(F, 0, Pending) of
              0 ->
                  io:format("No transfers active\n"),
                  ok;
              _ ->
                  error
          end
      catch
          Exception:Reason ->
              log_to_riak_node("transfers failed ~p:~p~n", [Exception,
                      Reason]),
              io:format("transfers failed, see log for details~n"),
              error
      end.
  
  
  cluster_info([OutFile|Rest]) ->
      try
          case lists:reverse(atomify_nodestrs(Rest)) of
              [] ->
                  cluster_info:dump_all_connected(OutFile);
              Nodes ->
                  cluster_info:dump_nodes(Nodes, OutFile)
          end
      catch
          error:{badmatch, {error, eacces}} ->
              io:format("cluster_info failed, permission denied writing to ~p~n", [OutFile]);
          error:{badmatch, {error, enoent}} ->
              io:format("cluster_info failed, no such directory ~p~n", [filename:dirname(OutFile)]);
          error:{badmatch, {error, enotdir}} ->
              io:format("cluster_info failed, not a directory ~p~n", [filename:dirname(OutFile)]);
          Exception:Reason ->
              log_to_riak_node("cluster_info failed ~p:~p~n",
                  [Exception, Reason]),
              io:format("cluster_info failed, see log for details~n"),
              error
      end.
  
  format_stats([], Acc) ->
  