%% ------------------------------------------------------------------- %% %% Copyright (c) 2011-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. %% %% ------------------------------------------------------------------- -module(riak_core_console). %% Legacy exports - unless needed by other modules, only expose %% functionality via command/1 -export([member_status/1, ring_status/1, print_member_status/1, stage_leave/1, stage_remove/1, stage_replace/1, stage_resize_ring/1, stage_force_replace/1, print_staged/1, commit_staged/1, clear_staged/1, transfer_limit/1, pending_claim_percentage/2, transfers/1, add_user/1, alter_user/1, del_user/1, add_group/1, alter_group/1, del_group/1, add_source/1, del_source/1, grant/1, revoke/1, print_users/1, print_user/1, print_sources/1, print_groups/1, print_group/1, print_grants/1, security_enable/1, security_disable/1, security_status/1, ciphers/1, timezone/1, stat_show/1, stat_info/1, stat_enable/1, stat_disable/1, stat_reset/1]). %% New CLI API -export([command/1]). -spec command([string()]) -> ok. command(Cmd) -> clique:run(Cmd). %% @doc Return for a given ring and node, percentage currently owned and %% anticipated after the transitions have been completed. pending_claim_percentage(Ring, Node) -> RingSize = riak_core_ring:num_partitions(Ring), FutureRingSize = riak_core_ring:future_num_partitions(Ring), Indices = riak_core_ring:indices(Ring, Node), NextIndices = riak_core_ring:future_indices(Ring, Node), RingPercent = length(Indices) * 100 / RingSize, NextPercent = length(NextIndices) * 100 / FutureRingSize, {RingPercent, NextPercent}. member_status([]) -> {ok, Ring} = riak_core_ring_manager:get_my_ring(), print_member_status(Ring). print_member_status(Ring) -> io:format("~33..=s Membership ~34..=s~n", ["", ""]), io:format("Status Ring Pending Node~n"), io:format("~79..-s~n", [""]), AllStatus = lists:keysort(2, riak_core_ring:all_member_status(Ring)), IsPending = ([] /= riak_core_ring:pending_changes(Ring)), {Joining, Valid, Down, Leaving, Exiting} = lists:foldl(fun({Node, Status}, {Joining0, Valid0, Down0, Leaving0, Exiting0}) -> {RingPercent, NextPercent} = pending_claim_percentage(Ring, Node), case IsPending of true -> io:format("~-8s ~5.1f% ~5.1f% ~p~n", [Status, RingPercent, NextPercent, Node]); false -> io:format("~-8s ~5.1f% -- ~p~n", [Status, RingPercent, Node]) end, case Status of joining -> {Joining0 + 1, Valid0, Down0, Leaving0, Exiting0}; valid -> {Joining0, Valid0 + 1, Down0, Leaving0, Exiting0}; down -> {Joining0, Valid0, Down0 + 1, Leaving0, Exiting0}; leaving -> {Joining0, Valid0, Down0, Leaving0 + 1, Exiting0}; exiting -> {Joining0, Valid0, Down0, Leaving0, Exiting0 + 1} end end, {0,0,0,0,0}, AllStatus), io:format("~79..-s~n", [""]), io:format("Valid:~b / Leaving:~b / Exiting:~b / Joining:~b / Down:~b~n", [Valid, Leaving, Exiting, Joining, Down]), ok. ring_status([]) -> {Claimant, RingReady, Down, MarkedDown, Changes} = riak_core_status:ring_status(), claimant_status(Claimant, RingReady), ownership_status(Down, Changes), unreachable_status(Down -- MarkedDown), ok. claimant_status(Claimant, RingReady) -> io:format("~34..=s Claimant ~35..=s~n", ["", ""]), io:format("Claimant: ~p~n", [Claimant]), case RingReady of undefined -> io:format("Status: down~n" "Ring Ready: unknown~n", []); _ -> io:format("Status: up~n" "Ring Ready: ~p~n", [RingReady]) end, io:format("~n", []). ownership_status(Down, Changes) -> io:format("~30..=s Ownership Handoff ~30..=s~n", ["", ""]), case Changes of [] -> io:format("No pending changes.~n"); _ -> orddict:fold(fun print_ownership_status/3, Down, Changes) end, io:format("~n", []). print_ownership_status({Owner, NextOwner}, Transfers, Down) -> io:format("Owner: ~s~n" "Next Owner: ~s~n", [Owner, NextOwner]), case {lists:member(Owner, Down), lists:member(NextOwner, Down)} of {true, true} -> io:format("~n"), io:format("!!! ~s is DOWN !!!~n", [Owner]), io:format("!!! ~s is DOWN !!!~n~n", [NextOwner]), lists:foreach(fun print_index/1, Transfers); {true, _} -> io:format("~n"), io:format("!!! ~s is DOWN !!!~n~n", [Owner]), lists:foreach(fun print_index/1, Transfers); {_, true} -> io:format("~n"), io:format("!!! ~s is DOWN !!!~n~n", [NextOwner]), lists:foreach(fun print_index/1, Transfers); _ -> lists:foreach(fun print_transfer_status/1, Transfers) end, io:format("~n"), io:format("~79..-s~n", [""]), Down. print_index({Idx, _Waiting, _Complete, _Status}) -> io:format("Index: ~b~n", [Idx]). print_transfer_status({Idx, Waiting, Complete, Status}) -> io:format("~nIndex: ~b~n", [Idx]), case Status of complete -> io:format(" All transfers complete. Waiting for " "claimant to change ownership.~n"); awaiting -> io:format(" Waiting on: ~p~n", [Waiting]), case Complete of [] -> ok; _ -> io:format(" Complete: ~p~n", [Complete]) end end. unreachable_status([]) -> io:format("~30..=s Unreachable Nodes ~30..=s~n", ["", ""]), io:format("All nodes are up and reachable~n", []), io:format("~n", []); unreachable_status(Down) -> io:format("~30..=s Unreachable Nodes ~30..=s~n", ["", ""]), io:format("The following nodes are unreachable: ~p~n", [Down]), io:format("~n", []), io:format("WARNING: The cluster state will not converge until all nodes~n" "are up. Once the above nodes come back online, convergence~n" "will continue. If the outages are long-term or permanent, you~n" "can either mark the nodes as down (riak-admin down NODE) or~n" "forcibly remove the nodes from the cluster (riak-admin~n" "force-remove NODE) to allow the remaining nodes to settle.~n"), ok. %% Provide a list of nodes with pending partition transfers (i.e. any secondary vnodes) %% and list any owned vnodes that are *not* running -spec(transfers([string()]) -> ok). transfers([]) -> try {DownNodes, Pending} = riak_core_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 -> lager:error("Transfers failed ~p:~p", [Exception, Reason]), io:format("Transfers failed, see log for details~n"), error end, %% Now display active transfers {Xfers, Down} = riak_core_status:all_active_transfers(), DisplayXfer = fun({{Mod, Partition}, Node, outbound, active, _Status}) -> print_v1_status(Mod, Partition, Node); ({status_v2, Status}) -> %% Display base status Type = proplists:get_value(type, Status), Mod = proplists:get_value(mod, Status), SrcPartition = proplists:get_value(src_partition, Status), TargetPartition = proplists:get_value(target_partition, Status), StartTS = proplists:get_value(start_ts, Status), SrcNode = proplists:get_value(src_node, Status), TargetNode = proplists:get_value(target_node, Status), print_v2_status(Type, Mod, {SrcPartition, TargetPartition}, StartTS), %% Get info about stats if there is any yet Stats = proplists:get_value(stats, Status), print_stats(SrcNode, TargetNode, Stats), io:format("~n"); (_) -> ignore end, DisplayDown = fun(Node) -> io:format("Node ~p could not be contacted~n", [Node]) end, io:format("~nActive Transfers:~n~n", []), _ = [DisplayXfer(Xfer) || Xfer <- lists:flatten(Xfers)], io:format("~n"), _ = [DisplayDown(Node) || Node <- Down], ok. print_v2_status(Type, Mod, {SrcPartition, TargetPartition}, StartTS) -> StartTSStr = datetime_str(StartTS), Running = timer:now_diff(os:timestamp(), StartTS), RunningStr = riak_core_format:human_time_fmt("~.2f", Running), io:format("transfer type: ~s~n", [Type]), io:format("vnode type: ~p~n", [Mod]), case Type of repair -> io:format("source partition: ~p~n", [SrcPartition]), io:format("target partition: ~p~n", [TargetPartition]); _ -> io:format("partition: ~p~n", [TargetPartition]) end, io:format("started: ~s [~s ago]~n", [StartTSStr, RunningStr]). print_v1_status(Mod, Partition, Node) -> io:format("vnode type: ~p~n", [Mod]), io:format("partition: ~p~n", [Partition]), io:format("target node: ~p~n~n", [Node]). print_stats(SrcNode, TargetNode, no_stats) -> io:format("last update: no updates seen~n"), print_size(undefined), io:format("objects transferred: unknown~n~n"), print_arrowbox(SrcNode, TargetNode, "unknown", "unknown", 0.0); print_stats(SrcNode, TargetNode, Stats) -> ObjsS = proplists:get_value(objs_per_s, Stats), BytesS = proplists:get_value(bytes_per_s, Stats), LastUpdate = proplists:get_value(last_update, Stats), Diff = timer:now_diff(os:timestamp(), LastUpdate), DiffStr = riak_core_format:human_time_fmt("~.2f", Diff), Objs = proplists:get_value(objs_total, Stats), ObjsSStr = riak_core_format:fmt("~p Objs/s", [ObjsS]), ByteStr = riak_core_format:human_size_fmt("~.2f", BytesS) ++ "/s", TS = datetime_str(LastUpdate), Size = proplists:get_value(size, Stats), DonePctDecimal = proplists:get_value(pct_done_decimal, Stats), io:format("last update: ~s [~s ago]~n", [TS, DiffStr]), print_size(Size), io:format("objects transferred: ~p~n~n", [Objs]), print_arrowbox(SrcNode, TargetNode, ObjsSStr, ByteStr, DonePctDecimal). print_size(undefined) -> io:format("total size: unknown~n"); print_size({Objs, objects}) -> io:format("total size: ~p objects~n", [Objs]); print_size({Bytes, bytes}) -> io:format("total size: ~p bytes~n", [Bytes]). -define(ARROW, "=======> "). print_arrowbox(SrcAtom, TargetAtom, Objs, Bytes, Progress) -> Src0 = atom_to_list(SrcAtom), Target0 = atom_to_list(TargetAtom), {SCont1, SCont2} = wrap(Src0), {TCont1, TCont2} = wrap(Target0), Src = case SCont1 of "" -> string:centre(Src0, 25); _ -> Src0 end, Target = case TCont1 of "" -> string:centre(Target0, 25); _ -> Target0 end, ToFrom = riak_core_format:fmt("~25s ~10s ~25s", [Src, ?ARROW, Target]), Width = length(ToFrom), Prog = progress(Progress, 50), io:format("~s~n", [string:centre(Objs, Width)]), io:format("~s~n", [ToFrom]), Fmt = "~-25s ~10s ~-25s~n", case SCont1 /= "" orelse TCont1 /= "" of true -> io:format(Fmt, [SCont1, "", TCont1]); _ -> ok end, case SCont2 /= "" orelse TCont2 /= "" of true -> io:format(Fmt, [SCont2, "", TCont2]); _ -> ok end, io:format("~s~n", [string:centre(" "++Prog, Width)]), io:format("~s~n", [string:centre(Bytes, Width)]). wrap(String) -> Len = length(String), case Len of N when N > 50 -> One = lists:sublist(String, 26, 25), Two = lists:sublist(String, 51, 25), {One, Two}; N when N >= 25, N =< 50-> One = lists:sublist(String, 26, 25), {One, ""}; _ -> {"", ""} end. progress(undefined, MaxSize) -> FormatStr = progress_fmt(progress_size(MaxSize), 0), %% this is wrong, need - 6 refactor riak_core_format:fmt(FormatStr, ["", "", "N/A"]); progress(PctDecimal, MaxSize) -> ProgressTotalSize = progress_size(MaxSize), ProgressSize = trunc(PctDecimal * ProgressTotalSize), PadSize = ProgressTotalSize - ProgressSize, FormatStr = progress_fmt(ProgressSize, PadSize), riak_core_format:fmt(FormatStr, ["", "", integer_to_list(trunc(PctDecimal * 100))]). progress_size(MaxSize) -> MaxSize - 7. %% 7 fixed characters in progress bar progress_fmt(ArrowSize, PadSize) -> riak_core_format:fmt("|~~~p..=s~~~p.. s| ~~3.. s%", [ArrowSize, PadSize]). datetime_str({_Mega, _Secs, _Micro}=Now) -> datetime_str(calendar:now_to_datetime(Now)); datetime_str({{Year, Month, Day}, {Hour, Min, Sec}}) -> riak_core_format:fmt("~4..0B-~2..0B-~2..0B ~2..0B:~2..0B:~2..0B", [Year,Month,Day,Hour,Min,Sec]). stage_leave([]) -> stage_leave(node()); stage_leave([NodeStr]) -> stage_leave(list_to_atom(NodeStr)); stage_leave(Node) -> try case riak_core_claimant:leave_member(Node) of ok -> io:format("Success: staged leave request for ~p~n", [Node]), ok; {error, already_leaving} -> io:format("~p is already in the process of leaving the " "cluster.~n", [Node]), ok; {error, not_member} -> io:format("Failed: ~p is not a member of the cluster.~n", [Node]), error; {error, only_member} -> io:format("Failed: ~p is the only member.~n", [Node]), error end catch Exception:Reason -> lager:error("Leave failed ~p:~p", [Exception, Reason]), io:format("Leave failed, see log for details~n"), error end. stage_remove([NodeStr]) -> stage_remove(list_to_atom(NodeStr)); stage_remove(Node) -> try case riak_core_claimant:remove_member(Node) of ok -> io:format("Success: staged remove request for ~p~n", [Node]), ok; {error, not_member} -> io:format("Failed: ~p is not a member of the cluster.~n", [Node]), error; {error, is_claimant} -> is_claimant_error(Node, "remove"), error; {error, only_member} -> io:format("Failed: ~p is the only member.~n", [Node]), error end catch Exception:Reason -> lager:error("Remove failed ~p:~p", [Exception, Reason]), io:format("Remove failed, see log for details~n"), error end. stage_replace([NodeStr1, NodeStr2]) -> stage_replace(list_to_atom(NodeStr1), list_to_atom(NodeStr2)). stage_replace(Node1, Node2) -> try case riak_core_claimant:replace(Node1, Node2) of ok -> io:format("Success: staged replacement of ~p with ~p~n", [Node1, Node2]), ok; {error, already_leaving} -> io:format("~p is already in the process of leaving the " "cluster.~n", [Node1]), ok; {error, not_member} -> io:format("Failed: ~p is not a member of the cluster.~n", [Node1]), error; {error, invalid_replacement} -> io:format("Failed: ~p is not a valid replacement candidate.~n" "Only newly joining nodes can be used for " "replacement.~n", [Node2]), error; {error, already_replacement} -> io:format("Failed: ~p is already staged to replace another " "node.~n", [Node2]), error end catch Exception:Reason -> lager:error("Node replacement failed ~p:~p", [Exception, Reason]), io:format("Node replacement failed, see log for details~n"), error end. stage_force_replace([NodeStr1, NodeStr2]) -> stage_force_replace(list_to_atom(NodeStr1), list_to_atom(NodeStr2)). stage_force_replace(Node1, Node2) -> try case riak_core_claimant:force_replace(Node1, Node2) of ok -> io:format("Success: staged forced replacement of ~p with ~p~n", [Node1, Node2]), ok; {error, not_member} -> io:format("Failed: ~p is not a member of the cluster.~n", [Node1]), error; {error, is_claimant} -> is_claimant_error(Node1, "replace"), error; {error, invalid_replacement} -> io:format("Failed: ~p is not a valid replacement candidate.~n" "Only newly joining nodes can be used for " "replacement.~n", [Node2]), error; {error, already_replacement} -> io:format("Failed: ~p is already staged to replace another " "node.~n", [Node2]), error end catch Exception:Reason -> lager:error("Forced node replacement failed ~p:~p", [Exception, Reason]), io:format("Forced node replacement failed, see log for details~n"), error end. stage_resize_ring(["abort"]) -> try case riak_core_claimant:abort_resize() of ok -> io:format("Success: staged abort resize ring request~n"), ok; {error, not_resizing} -> io:format("Failed: ring is not resizing or resize has completed~n"), error end catch Exception:Reason -> lager:error("Abort resize ring request failed ~p:~p", [Exception, Reason]), io:format("Abort resize ring request failed, see log for details~n"), error end; stage_resize_ring([SizeStr]) -> try list_to_integer(SizeStr) of Size -> stage_resize_ring(Size) catch error:badarg -> io:format("Failed: Ring size must be an integer.~n") end; stage_resize_ring(NewRingSize) -> try case riak_core_claimant:resize_ring(NewRingSize) of ok -> io:format("Success: staged resize ring request with new size: ~p~n", [NewRingSize]), ok; {error, not_capable} -> io:format("Failed: at least one node is not capable of performing the operation~n"), error; {error, same_size} -> io:format("Failed: current ring size is already ~p~n", [NewRingSize]), error; {error, control_running} -> io:format("Failed: cannot resize ring with Riak Control running~n"), error; {error, search_running} -> io:format("Failed: cannot resize ring with Riak Search~n"), error; {error, single_node} -> io:format("Failed: cannot resize single node~n"), error; {error, pending_changes} -> io:format("Failed: there are pending changes. Try again once completed~n"), error end catch Exception:Reason -> lager:error("Resize ring request failed ~p:~p", [Exception, Reason]), io:format("Resize ring request failed, see log for details~n"), error end. clear_staged([]) -> try case riak_core_claimant:clear() of ok -> io:format("Cleared staged cluster changes~n"), ok end catch Exception:Reason -> lager:error("Failed to clear staged cluster changes ~p:~p", [Exception, Reason]), io:format("Failed to clear staged cluster changes, see log " "for details~n"), error end. is_claimant_error(Node, Action) -> io:format("Failed: ~p is the claimant (see: riak-admin ring_status).~n", [Node]), io:format( "The claimant is the node responsible for initiating cluster changes,~n" "and cannot forcefully ~s itself. You can use 'riak-admin down' to~n" "mark the node as offline, which will trigger a new claimant to take~n" "over. However, this will clear any staged changes.~n", [Action]). print_staged([]) -> case riak_core_claimant:plan() of {error, ring_not_ready} -> io:format("Cannot plan until cluster state has converged.~n" "Check 'Ring Ready' in 'riak-admin ring_status'~n"); {error, invalid_resize_claim} -> io:format("Unable to claim some partitions in resized ring.~n" "Check that there are no pending changes in 'riak-admin ring-status'~n" "If there are, try again once they are completed~n" "Otherwise try again shortly.~n"); {ok, Changes, NextRings} -> {ok, Ring} = riak_core_ring_manager:get_my_ring(), %% The last next ring is always the final ring after all changes, %% which is uninteresting to show. Only print N-1 rings. NextRings2 = lists:sublist(NextRings, erlang:max(0, length(NextRings)-1)), print_plan(Changes, Ring, NextRings2), ok end. print_plan([], _, _) -> io:format("There are no staged changes~n"); print_plan(Changes, Ring, NextRings) -> io:format("~31..=s Staged Changes ~32..=s~n", ["", ""]), io:format("Action Details(s)~n"), io:format("~79..-s~n", [""]), lists:foreach(fun({Node, join}) -> io:format("join ~p~n", [Node]); ({Node, leave}) -> io:format("leave ~p~n", [Node]); ({Node, remove}) -> io:format("force-remove ~p~n", [Node]); ({Node, {replace, NewNode}}) -> io:format("replace ~p with ~p~n", [Node, NewNode]); ({Node, {force_replace, NewNode}}) -> io:format("force-replace ~p with ~p~n", [Node, NewNode]); ({_, {resize, NewRingSize}}) -> CurrentSize = riak_core_ring:num_partitions(Ring), io:format("resize-ring ~p to ~p partitions~n",[CurrentSize,NewRingSize]); ({_, abort_resize}) -> CurrentSize = riak_core_ring:num_partitions(Ring), io:format("resize-ring abort. current size: ~p~n", [CurrentSize]) end, Changes), io:format("~79..-s~n", [""]), io:format("~n"), lists:foreach(fun({Node, remove}) -> io:format("WARNING: All of ~p replicas will be lost~n", [Node]); ({Node, {force_replace, _}}) -> io:format("WARNING: All of ~p replicas will be lost~n", [Node]); (_) -> ok end, Changes), io:format("~n"), Transitions = length(NextRings), case Transitions of 1 -> io:format("NOTE: Applying these changes will result in 1 " "cluster transition~n~n"); _ -> io:format("NOTE: Applying these changes will result in ~b " "cluster transitions~n~n", [Transitions]) end, _ = lists:foldl(fun({Ring1, Ring2}, I) -> io:format("~79..#s~n", [""]), io:format("~24.. s After cluster transition ~b/~b~n", ["", I, Transitions]), io:format("~79..#s~n~n", [""]), output(Ring1, Ring2), I+1 end, 1, NextRings), ok. output(Ring, NextRing) -> riak_core_console:print_member_status(NextRing), io:format("~n"), FutureRing = riak_core_ring:future_ring(NextRing), case riak_core_ring_util:check_ring(FutureRing) of [] -> ok; _ -> io:format("WARNING: Not all replicas will be on distinct nodes~n~n") end, Owners1 = riak_core_ring:all_owners(Ring), Owners2 = riak_core_ring:all_owners(NextRing), Owners3 = lists:zip(Owners1, Owners2), Reassigned = [{Idx, PrevOwner, NewOwner} || {{Idx, PrevOwner}, {Idx, NewOwner}} <- Owners3, PrevOwner /= NewOwner], ReassignedTally = tally(Reassigned), Pending = riak_core_ring:pending_changes(NextRing), Next = [{Idx, PrevOwner, NewOwner} || {Idx, PrevOwner, NewOwner, _, _} <- Pending], NextTally = tally(Next), Resizing = riak_core_ring:is_resizing(NextRing), case Reassigned of [] -> ok; _ -> io:format("Partitions reassigned from cluster changes: ~p~n", [length(Reassigned)]), _ = [io:format(" ~b reassigned from ~p to ~p~n", [Count, PrevOwner, NewOwner]) || {{PrevOwner, NewOwner}, Count} <- ReassignedTally], io:format("~n"), ok end, case {Resizing, Next} of {_, []} -> ok; {true, _} -> io:format("Ring is resizing. see riak-admin ring-status for transfer details.~n"); _ -> io:format("Transfers resulting from cluster changes: ~p~n", [length(Next)]), _ = [io:format(" ~b transfers from ~p to ~p~n", [Count, PrevOwner, NewOwner]) || {{PrevOwner, NewOwner}, Count} <- NextTally], ok, io:format("~n") end, ok. tally(Changes) -> Tally = lists:foldl(fun({_, PrevOwner, NewOwner}, Tally) -> dict:update_counter({PrevOwner, NewOwner}, 1, Tally) end, dict:new(), Changes), dict:to_list(Tally). commit_staged([]) -> case riak_core_claimant:commit() of ok -> io:format("Cluster changes committed~n"); {error, nothing_planned} -> io:format("You must verify the plan with " "'riak-admin cluster plan' before committing~n"); {error, ring_not_ready} -> io:format("Cannot commit until cluster state has converged.~n" "Check 'Ring Ready' in 'riak-admin ring_status'~n"); {error, plan_changed} -> io:format("The plan has changed. Verify with " "'riak-admin cluster plan' before committing~n"); {error, invalid_resize_claim} -> io:format("Unable to commit staged ring changes.~n" "Check that there are no pending changes in 'riak-admin ring-status'~n" "If there are, try again once they are completed,~n" "Otherwise try again shortly.~n"); _ -> io:format("Unable to commit cluster changes. Plan " "may have changed, please verify the~n" "plan and try to commit again~n") end. timezone([]) -> TZ = riak_core_metadata:get({riak_core, timezone}, string, [{default, "not configured"}]), io:format("~s~n", [TZ]); timezone([TZString]) -> TZ = jam:compile(jam_iso8601:parse_tz(TZString)), attempt_set_timezone(jam:is_valid(TZ), TZ, TZString). attempt_set_timezone(false, _TZ, String) -> io:format("Failed, unrecognized timezone string '~s'~n", [String]); attempt_set_timezone(true, TZ, _String) -> NewString = jam_iso8601:to_string(TZ, [{z, false}]), riak_core_metadata:put({riak_core, timezone}, string, NewString), riak_core_metadata:put({riak_core, timezone}, seconds, jam:tz_to_seconds(TZ)), io:format("Success, timezone is now configured to '~s'~n", [NewString]). transfer_limit([]) -> {Limits, Down} = riak_core_util:rpc_every_member_ann(riak_core_handoff_manager, get_concurrency, [], 5000), io:format("~s~n", [string:centre(" Transfer Limit ", 79, $=)]), io:format("Limit Node~n"), io:format("~79..-s~n", [""]), lists:foreach(fun({Node, Limit}) -> io:format("~5b ~p~n", [Limit, Node]) end, Limits), lists:foreach(fun(Node) -> io:format("(offline) ~p~n", [Node]) end, Down), io:format("~79..-s~n", [""]), io:format("Note: You can change transfer limits with " "'riak-admin transfer_limit '~n" " and 'riak-admin transfer_limit '~n"), ok; transfer_limit([LimitStr]) -> {Valid, Limit} = check_limit(LimitStr), case Valid of false -> io:format("Invalid limit: ~s~n", [LimitStr]), error; true -> io:format("Setting transfer limit to ~b across the cluster~n", [Limit]), {_, Down} = riak_core_util:rpc_every_member_ann(riak_core_handoff_manager, set_concurrency, [Limit], 5000), (Down == []) orelse io:format("Failed to set limit for: ~p~n", [Down]), ok end; transfer_limit([NodeStr, LimitStr]) -> Node = list_to_atom(NodeStr), {Valid, Limit} = check_limit(LimitStr), case Valid of false -> io:format("Invalid limit: ~s~n", [LimitStr]), error; true -> case riak_core_util:safe_rpc(Node, riak_core_handoff_manager, set_concurrency, [Limit]) of {badrpc, _} -> io:format("Failed to set transfer limit for ~p~n", [Node]); _ -> io:format("Set transfer limit for ~p to ~b~n", [Node, Limit]) end, ok end. check_limit(Str) -> try Int = list_to_integer(Str), {Int >= 0, Int} catch _:_ -> {false, 0} end. security_error_xlate({errors, Errors}) -> string:join( lists:map(fun(X) -> security_error_xlate({error, X}) end, Errors), "~n"); security_error_xlate({error, unknown_user}) -> "User not recognized"; security_error_xlate({error, unknown_group}) -> "Group not recognized"; security_error_xlate({error, {unknown_permission, Name}}) -> io_lib:format("Permission not recognized: ~ts", [Name]); security_error_xlate({error, {unknown_role, Name}}) -> io_lib:format("Name not recognized: ~ts", [Name]); security_error_xlate({error, {unknown_user, Name}}) -> io_lib:format("User not recognized: ~ts", [Name]); security_error_xlate({error, {unknown_group, Name}}) -> io_lib:format("Group not recognized: ~ts", [Name]); security_error_xlate({error, {unknown_users, Names}}) -> io_lib:format("User(s) not recognized: ~ts", [ string:join( lists:map(fun(X) -> unicode:characters_to_list(X, utf8) end, Names), ", ") ]); security_error_xlate({error, {unknown_groups, Names}}) -> io_lib:format("Group(s) not recognized: ~ts", [ string:join( lists:map(fun(X) -> unicode:characters_to_list(X, utf8) end, Names), ", ") ]); security_error_xlate({error, {unknown_roles, Names}}) -> io_lib:format("Name(s) not recognized: ~ts", [ string:join( lists:map(fun(X) -> unicode:characters_to_list(X, utf8) end, Names), ", ") ]); security_error_xlate({error, {duplicate_roles, Names}}) -> io_lib:format("Ambiguous names need to be prefixed with 'user/' or 'group/': ~ts", [ string:join( lists:map(fun(X) -> unicode:characters_to_list(X, utf8) end, Names), ", ") ]); security_error_xlate({error, reserved_name}) -> "This name is reserved for system use"; security_error_xlate({error, no_matching_sources}) -> "No matching source"; security_error_xlate({error, illegal_name_char}) -> "Illegal character(s) in name"; security_error_xlate({error, role_exists}) -> "This name is already in use"; %% If we get something we hadn't planned on, better an ugly error %% message than an ugly RPC call failure security_error_xlate(Error) -> io_lib:format("~p", [Error]). add_user([Username|Options]) -> add_role(Username, Options, fun riak_core_security:add_user/2). add_group([Groupname|Options]) -> add_role(Groupname, Options, fun riak_core_security:add_group/2). add_role(Name, Options, Fun) -> try Fun(Name, parse_options(Options)) of ok -> ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error catch throw:{error, {invalid_option, Option}} -> io:format("Invalid option ~p, options are of the form key=value~n", [Option]), error end. alter_user([Username|Options]) -> alter_role(Username, Options, fun riak_core_security:alter_user/2). alter_group([Groupname|Options]) -> alter_role(Groupname, Options, fun riak_core_security:alter_group/2). alter_role(Name, Options, Fun) -> try Fun(Name, parse_options(Options)) of ok -> ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error catch throw:{error, {invalid_option, Option}} -> io:format("Invalid option ~p, options are of the form key=value~n", [Option]), error end. del_user([Username]) -> del_role(Username, fun riak_core_security:del_user/1). del_group([Groupname]) -> del_role(Groupname, fun riak_core_security:del_group/1). del_role(Name, Fun) -> case Fun(Name) of ok -> io:format("Successfully deleted ~ts~n", [Name]), ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error end. add_source([Users, CIDR, Source | Options]) -> Unames = case string:tokens(Users, ",") of ["all"] -> all; Other -> Other end, %% Unicode note: atoms are constrained to latin1 until R18, so our %% sources are as well try riak_core_security:add_source(Unames, parse_cidr(CIDR), list_to_atom(string:to_lower(Source)), parse_options(Options)) of ok -> io:format("Successfully added source~n"), ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error catch throw:{error, {invalid_option, Option}} -> io:format("Invalid option ~p, options are of the form key=value~n", [Option]); error:badarg -> io:format("Invalid source ~ts, must be latin1, sorry~n", [Source]) end. del_source([Users, CIDR]) -> Unames = case string:tokens(Users, ",") of ["all"] -> all; Other -> Other end, riak_core_security:del_source(Unames, parse_cidr(CIDR)), io:format("Deleted source~n"). parse_roles(Roles) -> case string:tokens(Roles, ",") of ["all"] -> all; Other -> Other end. parse_grants(Grants) -> string:tokens(Grants, ","). grant_int(Permissions, Bucket, Roles) -> case riak_core_security:add_grant(Roles, Bucket, Permissions) of ok -> io:format("Successfully granted~n"), ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error end. grant([Grants, "on", "any", "to", Users]) -> grant_int(parse_grants(Grants), any, parse_roles(Users)); grant([Grants, "on", Type, Bucket, "to", Users]) -> grant_int(parse_grants(Grants), { Type, Bucket }, parse_roles(Users)); grant([Grants, "on", Type, "to", Users]) -> grant_int(parse_grants(Grants), Type, parse_roles(Users)); grant(_) -> io:format("Usage: grant on ( [bucket]|any) to ~n"), error. revoke_int(Permissions, Bucket, Roles) -> case riak_core_security:add_revoke(Roles, Bucket, Permissions) of ok -> io:format("Successfully revoked~n"), ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error end. revoke([Grants, "on", "any", "from", Users]) -> revoke_int(parse_grants(Grants), any, parse_roles(Users)); revoke([Grants, "on", Type, Bucket, "from", Users]) -> revoke_int(parse_grants(Grants), { Type, Bucket }, parse_roles(Users)); revoke([Grants, "on", Type, "from", Users]) -> revoke_int(parse_grants(Grants), Type, parse_roles(Users)); revoke(_) -> io:format("Usage: revoke on [bucket] from ~n"), error. print_grants([Name]) -> case riak_core_security:print_grants(Name) of ok -> ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error end. print_users([]) -> riak_core_security:print_users(). print_user([User]) -> case riak_core_security:print_user(User) of ok -> ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error end. print_groups([]) -> riak_core_security:print_groups(). print_group([Group]) -> case riak_core_security:print_group(Group) of ok -> ok; Error -> io:format(security_error_xlate(Error)), io:format("~n"), Error end. print_sources([]) -> riak_core_security:print_sources(). ciphers([]) -> riak_core_security:print_ciphers(); ciphers([CipherList]) -> case riak_core_security:set_ciphers(CipherList) of ok -> riak_core_security:print_ciphers(), ok; error -> error end. security_enable([]) -> riak_core_security:enable(). security_disable([]) -> riak_core_security:disable(). security_status([]) -> case riak_core_security:status() of enabled -> io:format("Enabled~n"); disabled -> io:format("Disabled~n"); enabled_but_no_capability -> io:format("WARNING: Configured to be enabled, but not supported " "on all nodes so it is disabled!~n") end. parse_options(Options) -> parse_options(Options, []). parse_options([], Acc) -> Acc; parse_options([H|T], Acc) -> case re:split(H, "=", [{parts, 2}, {return, list}]) of [Key, Value] when is_list(Key), is_list(Value) -> parse_options(T, [{string:to_lower(Key), Value}|Acc]); _Other -> throw({error, {invalid_option, H}}) end. -spec parse_cidr(string()) -> {inet:ip_address(), non_neg_integer()}. parse_cidr(CIDR) -> [IP, Mask] = string:tokens(CIDR, "/"), {ok, Addr} = inet_parse:address(IP), {Addr, list_to_integer(Mask)}. stat_show(Arg) -> print_stats(find_entries(Arg)). find_entries(Arg) -> find_entries(Arg, enabled). find_entries(Arg, Status0) -> lists:map( fun(A) -> {S, Type, Status, DPs} = type_status_and_dps(A, Status0), case S of "[" ++ _ -> {find_entries_1(S, Type, Status), DPs}; _ -> case legacy_search(S, Type, Status) of false -> {find_entries_1(S, Type, Status), DPs}; Found -> {Found, DPs} end end end, Arg). type_status_and_dps(S, Status0) -> [S1|Rest] = re:split(S, "/"), {Type, Status, DPs} = type_status_and_dps(Rest, '_', Status0, default), {S1, Type, Status, DPs}. type_status_and_dps([<<"type=", T/binary>>|Rest], _Type, Status, DPs) -> NewType = case T of <<"*">> -> '_'; _ -> try binary_to_existing_atom(T, latin1) catch error:_ -> T end end, type_status_and_dps(Rest, NewType, Status, DPs); type_status_and_dps([<<"status=", St/binary>>|Rest], Type, _Status, DPs) -> NewStatus = case St of <<"enabled">> -> enabled; <<"disabled">> -> disabled; <<"*">> -> '_' end, type_status_and_dps(Rest, Type, NewStatus, DPs); type_status_and_dps([DPsBin|Rest], Type, Status, DPs) -> NewDPs = merge([binary_to_existing_atom(D,latin1) || D <- re:split(DPsBin, ",")], DPs), type_status_and_dps(Rest, Type, Status, NewDPs); type_status_and_dps([], Type, Status, DPs) -> {Type, Status, DPs}. merge([_|_] = DPs, default) -> DPs; merge([H|T], DPs) -> case lists:member(H, DPs) of true -> merge(T, DPs); false -> merge(T, DPs ++ [H]) end; merge([], DPs) -> DPs. find_entries_1(S, Type, Status) -> Patterns = lists:flatten([parse_stat_entry(S, Type, Status)]), exometer:select(Patterns). legacy_search(S, Type, Status) -> case re:run(S, "\\.", []) of {match,_} -> false; nomatch -> Re = <<"^", (make_re(S))/binary, "$">>, [{S, legacy_search_1(Re, Type, Status)}] end. make_re(S) -> repl(split_pattern(S, [])). repl([single|T]) -> <<"[^_]*", (repl(T))/binary>>; repl([double|T]) -> <<".*", (repl(T))/binary>>; repl([H|T]) -> <>; repl([]) -> <<>>. split_pattern(<<>>, Acc) -> lists:reverse(Acc); split_pattern(<<"**", T/binary>>, Acc) -> split_pattern(T, [double|Acc]); split_pattern(<<"*", T/binary>>, Acc) -> split_pattern(T, [single|Acc]); split_pattern(B, Acc) -> case binary:match(B, <<"*">>) of {Pos,_} -> <> = B, split_pattern(Rest, [Bef|Acc]); nomatch -> lists:reverse([B|Acc]) end. legacy_search_1(N, Type, Status) -> Found = exometer_alias:regexp_foldr(N, fun(Alias, Entry, DP, Acc) -> orddict:append(Entry, {DP,Alias}, Acc) end, orddict:new()), lists:foldr( fun({Entry, DPs}, Acc) -> case match_type(Entry, Type) of true -> DPnames = [D || {D,_} <- DPs], case exometer:get_value(Entry, DPnames) of {ok, Values} when is_list(Values) -> [{Entry, zip_values(Values, DPs)} | Acc]; {ok, disabled} when Status=='_'; Status==disabled -> [{Entry, zip_disabled(DPs)} | Acc]; _ -> [{Entry, [{D,undefined} || D <- DPnames]}|Acc] end; false -> Acc end end, [], orddict:to_list(Found)). match_type(_, '_') -> true; match_type(Name, T) -> T == exometer:info(Name, type). zip_values([{D,V}|T], DPs) -> {_,N} = lists:keyfind(D, 1, DPs), [{D,V,N}|zip_values(T, DPs)]; zip_values([], _) -> []. zip_disabled(DPs) -> [{D,disabled,N} || {D,N} <- DPs]. print_stats([]) -> io:fwrite("No matching stats~n", []); print_stats(Entries) -> lists:foreach( fun({[{LP, []}], _}) -> io:fwrite( "== ~s (Legacy pattern): No matching stats ==~n", [LP]); ({[{LP, Matches}], _}) -> io:fwrite("== ~s (Legacy pattern): ==~n", [LP]), [[io:fwrite("~p: ~p (~p/~p)~n", [N, V, E, DP]) || {DP,V,N} <- DPs] || {E, DPs} <- Matches]; ({[], _}) -> io:fwrite("No matching stats~n", []); ({Entries1, DPs}) -> [io:fwrite("~p: ~p~n", [E, get_value(E, Status, DPs)]) || {E, _, Status} <- Entries1] end, Entries). get_value(_, disabled, _) -> disabled; get_value(E, _Status, DPs) -> case exometer:get_value(E, DPs) of {ok, V} -> V; {error,_} -> unavailable end. stat_change(Arg, ToStatus) -> lists:foreach( fun({[{LP, []}], _}) -> io:fwrite( "== ~s (Legacy pattern): No matching stats ==~n", [LP]); ({[{LP, Matches}], _}) -> io:fwrite("== ~s (Legacy pattern): ==~n", [LP]), [io:fwrite("~p: ~p~n", [N, change_status(N, ToStatus)]) || {N, _} <- Matches]; ({[], _}) -> io:fwrite("No matching stats~n", []); ({Entries, _}) -> [io:fwrite("~p: ~p~n", [N, change_status(N, ToStatus)]) || {N, _, _} <- Entries] end, find_entries(Arg, '_')). stat_enable(Arg) -> stat_change(Arg, enabled). stat_disable(Arg) -> stat_change(Arg, disabled). stat_reset(Arg) -> lists:foreach( fun({[{LP, []}], _}) -> io:fwrite( "== ~s (Legacy pattern): No matching stats ==~n", [LP]); ({[{LP, Matches}], _}) -> io:fwrite("== ~s (Legacy pattern): ==~n", [LP]), [io:fwrite("~p: ~p~n", [N, exometer:reset(N)]) || {N, _} <- Matches]; ({Entries, _}) -> [io:fwrite("~p: ~p~n", [N, exometer:reset(N)]) || {N, _, _} <- Entries] end, find_entries(Arg, enabled)). change_status(N, St) -> case exometer:setopts(N, [{status, St}]) of ok -> St; Error -> Error end. stat_info(Arg) -> {Attrs, RestArg} = pick_info_attrs(split_arg(Arg)), [print_info(E, Attrs) || E <- find_entries(RestArg, '_')]. pick_info_attrs(Arg) -> case lists:foldr( fun("-name" , {As, Ps}) -> {[name |As], Ps}; ("-type" , {As, Ps}) -> {[type |As], Ps}; ("-module" , {As, Ps}) -> {[module |As], Ps}; ("-value" , {As, Ps}) -> {[value |As], Ps}; ("-cache" , {As, Ps}) -> {[cache |As], Ps}; ("-status" , {As, Ps}) -> {[status |As], Ps}; ("-timestamp", {As, Ps}) -> {[timestamp|As], Ps}; ("-options" , {As, Ps}) -> {[options |As], Ps}; (P, {As, Ps}) -> {As, [P|Ps]} end, {[], []}, Arg) of {[], Rest} -> {[name, type, module, value, cache, status, timestamp, options], Rest}; Other -> Other end. print_info({[{LP, []}], _}, _) -> io:fwrite("== ~s (Legacy pattern): No matching stats ==~n", [LP]); print_info({[{LP, Matches}], _}, Attrs) -> io:fwrite("== ~s (Legacy pattern): ==~n", [LP]), lists:foreach( fun({N, _}) -> print_info_1(N, Attrs) end, Matches); print_info({[], _}, _) -> io_lib:fwrite("No matching stats~n", []); print_info({Entries, _}, Attrs) -> lists:foreach( fun({N,_,_}) -> print_info_1(N, Attrs) end, Entries). print_info_1(N, [A|Attrs]) -> Hdr = lists:flatten(io_lib:fwrite("~p: ", [N])), Pad = lists:duplicate(length(Hdr), $\s), Info = exometer:info(N), Status = proplists:get_value(status, Info, enabled), Body = [io_lib:fwrite("~w = ~p~n", [A, proplists:get_value(A, Info)]) | lists:map(fun(value) -> io_lib:fwrite(Pad ++ "~w = ~p~n", [value, get_value(N, Status, default)]); (Ax) -> io_lib:fwrite(Pad ++ "~w = ~p~n", [Ax, proplists:get_value(Ax, Info)]) end, Attrs)], io:put_chars([Hdr, Body]). split_arg([Str]) -> re:split(Str, "\\s", [{return,list}]). parse_stat_entry([], Type, Status) -> {{[riak_core_stat:prefix()] ++ '_', Type, '_'}, [{'=:=','$status',Status}], ['$_']}; parse_stat_entry("*", Type, Status) -> parse_stat_entry([], Type, Status); parse_stat_entry("[" ++ _ = Expr, _Type, _Status) -> case erl_scan:string(ensure_trailing_dot(Expr)) of {ok, Toks, _} -> case erl_parse:parse_exprs(Toks) of {ok, [Abst]} -> partial_eval(Abst); Error -> io:fwrite("(Parse error for ~p: ~p~n", [Expr, Error]), [] end; ScanErr -> io:fwrite("(Scan error for ~p: ~p~n", [Expr, ScanErr]), [] end; parse_stat_entry(Str, Type, Status) when Status==enabled; Status==disabled -> Parts = re:split(Str, "\\.", [{return,list}]), Heads = replace_parts(Parts), [{{H,Type,Status}, [], ['$_']} || H <- Heads]; parse_stat_entry(Str, Type, '_') -> Parts = re:split(Str, "\\.", [{return,list}]), Heads = replace_parts(Parts), [{{H,Type,'_'}, [], ['$_']} || H <- Heads]; parse_stat_entry(_, _, Status) -> io:fwrite("(Illegal status: ~p~n", [Status]). ensure_trailing_dot(Str) -> case lists:reverse(Str) of "." ++ _ -> Str; _ -> Str ++ "." end. partial_eval({cons,_,H,T}) -> [partial_eval(H) | partial_eval(T)]; %% partial_eval({nil,_}) -> %% []; partial_eval({tuple,_,Elems}) -> list_to_tuple([partial_eval(E) || E <- Elems]); %% partial_eval({T,_,X}) when T==atom; T==integer; T==float -> %% X; partial_eval({op,_,'++',L1,L2}) -> partial_eval(L1) ++ partial_eval(L2); partial_eval(X) -> erl_parse:normalise(X). replace_parts(Parts) -> case split("**", Parts) of {_, []} -> [replace_parts_1(Parts)]; {Before, After} -> Head = replace_parts_1(Before), Tail = replace_parts_1(After), [Head ++ Pad ++ Tail || Pad <- pads()] end. pads() -> [['_'], ['_','_'], ['_','_','_'], ['_','_','_','_'], ['_','_','_','_','_'], ['_','_','_','_','_','_'], ['_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_','_','_','_','_','_'], ['_','_','_','_','_','_','_','_','_','_','_','_','_','_','_','_']]. split(X, L) -> split(L, X, []). split([H|T], H, Acc) -> {lists:reverse(Acc), T}; split([H|T], X, Acc) -> split(T, X, [H|Acc]); split([], _, Acc) -> {lists:reverse(Acc), []}. replace_parts_1([H|T]) -> R = replace_part(H), case T of ["**"] -> [R] ++ '_'; _ -> [R|replace_parts_1(T)] end; replace_parts_1([]) -> []. replace_part(H) -> case H of "*" -> '_'; "'" ++ _ -> case erl_scan:string(H) of {ok, [{atom, _, A}], _} -> A; Error -> error(Error) end; [C|_] when C >= $0, C =< $9 -> try list_to_integer(H) catch error:_ -> list_to_atom(H) end; _ -> list_to_atom(H) end.