anchor.diff:  %% and list any owned vnodes that are *not* running
   -spec(transfers([]) -> ok).
   transfers([]) ->
  -    {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) ->
  +    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) ->
  +            ({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.
  +        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]) ->
  -    case lists:reverse(atomify_nodestrs(Rest)) of
  -        [] ->
  -            cluster_info:dump_all_connected(OutFile);
  -        Nodes ->
  -            cluster_info:dump_nodes(Nodes, OutFile)
  +    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) ->
  
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:          cluster_info/1]).
  
  join([NodeStr]) ->
      try
          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
      catch
          Exception:Reason ->
              log_to_riak_node("join failed ~p:~p~n", [Exception,
                      Reason]),
              io:format("join failed, see log for details~n"),
              error
      end.
  
  
  leave([]) ->
      remove_node(node()).
  