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:          cluster_info/1]).
  
  join([NodeStr]) ->
      case riak:join(NodeStr) of
          ok ->
              io:format("Sent join request to ~s\n", [NodeStr]),
              ok;
          {error, not_reachable} ->
              io:format("Node ~s is not reachable!\n", [NodeStr]),
              error;
          {error, different_ring_sizes} ->
              io:format("Failed: ~s has a different ring_creation_size~n",
                        [NodeStr]),
              error
      end;
  join(_) ->
      io:format("Join requires a node to join with.\n"),
      error.
  
  leave([]) ->
      remove_node(node()).
  
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) ->
  