X-Git-Url: https://git.xandkar.net/?a=blobdiff_plain;f=src%2Fbeam_stats_msg_graphite.erl;h=73b770c35d541846de571b0677861c7c25b9026c;hb=e86061d0f13b72a2ea04a96c5c0a6585b58a7beb;hp=c88e3270d0c79c4363f45fc9c2f575581b52876f;hpb=ece99ea366e3bc8720528aaf5e1e1563e85b561c;p=beam_stats.git diff --git a/src/beam_stats_msg_graphite.erl b/src/beam_stats_msg_graphite.erl index c88e327..73b770c 100644 --- a/src/beam_stats_msg_graphite.erl +++ b/src/beam_stats_msg_graphite.erl @@ -14,12 +14,12 @@ -export( [ of_beam_stats/1 , of_beam_stats/2 - , to_bin/1 - , path_to_bin/1 - , add_path_prefix/2 + , to_iolist/1 + , path_to_iolist/1 , node_id_to_bin/1 ]). +-define(SCHEMA_VERSION, <<"beam_stats_v0">>). -define(T, #?MODULE). -type t() :: @@ -41,7 +41,6 @@ of_beam_stats(#beam_stats { timestamp = Timestamp , node_id = _ , memory = Memory - % TODO: Handle the rest of data points , io_bytes_in = IOBytesIn , io_bytes_out = IOBytesOut , context_switches = ContextSwitches @@ -54,40 +53,37 @@ of_beam_stats(#beam_stats ) -> Ts = Timestamp, N = NodeID, - [ cons([N, <<"io">> , <<"bytes_in">> ], IOBytesIn , Ts) - , cons([N, <<"io">> , <<"bytes_out">>], IOBytesOut , Ts) - , cons([N, <<"context_switches">> ], ContextSwitches, Ts) - , cons([N, <<"reductions">> ], Reductions , Ts) - , cons([N, <<"run_queue">> ], RunQueue , Ts) - | of_memory(Memory, NodeID, Ts) - ] - ++ of_ets(ETS, NodeID, Ts) - ++ of_processes(Processes, NodeID, Ts). + Msgs = + [ cons([N, <<"io">> , <<"bytes_in">> ], IOBytesIn , Ts) + , cons([N, <<"io">> , <<"bytes_out">>], IOBytesOut , Ts) + , cons([N, <<"context_switches">> ], ContextSwitches, Ts) + , cons([N, <<"reductions">> ], Reductions , Ts) + , cons([N, <<"run_queue">> ], RunQueue , Ts) + | of_memory(Memory, NodeID, Ts) + ] + ++ of_ets(ETS, NodeID, Ts) + ++ of_processes(Processes, NodeID, Ts), + lists:map(fun path_prefix_schema_version/1, Msgs). --spec to_bin(t()) -> - binary(). -to_bin( +-spec to_iolist(t()) -> + iolist(). +to_iolist( ?T { path = Path , value = Value , timestamp = Timestamp } ) -> - PathBin = path_to_bin(Path), + PathIOList = path_to_iolist(Path), ValueBin = integer_to_binary(Value), TimestampInt = timestamp_to_integer(Timestamp), TimestampBin = integer_to_binary(TimestampInt), - <>. - --spec add_path_prefix(t(), binary()) -> - t(). -add_path_prefix(?T{path=Path}=T, <>) -> - T?T{path = [Prefix | Path]}. + [PathIOList, <<" ">>, ValueBin, <<" ">>, TimestampBin]. --spec path_to_bin([binary()]) -> - binary(). -path_to_bin(Path) -> - bin_join(Path, <<".">>). +-spec path_to_iolist([binary()]) -> + iolist(). +path_to_iolist(Path) -> + list_interleave(Path, <<".">>). -spec node_id_to_bin(node()) -> binary(). @@ -99,13 +95,24 @@ node_id_to_bin(NodeID) -> %% Helpers %% ============================================================================ --spec bin_join([binary()], binary()) -> - binary(). -bin_join([] , <<_/binary>> ) -> <<>>; -bin_join([<> | []] , <<_/binary>> ) -> B; -bin_join([<> | [_|_]=Bins], <>) -> - BinsBin = bin_join(Bins, Sep), - <>. +-spec path_prefix_schema_version(t()) -> + t(). +path_prefix_schema_version(?T{}=T) -> + path_prefix(T, ?SCHEMA_VERSION). + +-spec path_prefix(t(), binary()) -> + t(). +path_prefix(?T{path=Path}=T, <>) -> + T?T{path = [Prefix | Path]}. + +-spec list_interleave([A], A) -> + [A]. +list_interleave([], _) -> + []; +list_interleave([X], _) -> + [X]; +list_interleave([X|Xs], Sep) -> + [X, Sep | list_interleave(Xs, Sep)]. -spec timestamp_to_integer(erlang:timestamp()) -> non_neg_integer(). @@ -122,12 +129,13 @@ of_memory(Memory, <>, Timestamp) -> end, lists:map(ComponentToMessage, Memory). --spec of_ets(beam_stats_ets_table:t(), binary(), erlang:timestamp()) -> +-spec of_ets(beam_stats_ets:t(), binary(), erlang:timestamp()) -> [t()]. of_ets(PerTableStats, <>, Timestamp) -> OfEtsTable = fun (Table) -> of_ets_table(Table, NodeID, Timestamp) end, - NestedMsgs = lists:map(OfEtsTable, PerTableStats), - lists:append(NestedMsgs). + MsgsNested = lists:map(OfEtsTable, PerTableStats), + MsgsFlattened = lists:append(MsgsNested), + aggregate_by_path(MsgsFlattened, Timestamp). -spec of_ets_table(beam_stats_ets_table:t(), binary(), erlang:timestamp()) -> [t()]. @@ -140,9 +148,13 @@ of_ets_table(#beam_stats_ets_table <>, Timestamp ) -> - IDBin = beam_stats_ets_table:id_to_bin(ID), + IDType = + case ID =:= Name + of true -> <<"NAMED">> + ; false -> <<"TID">> + end, NameBin = atom_to_binary(Name, latin1), - NameAndID = [NameBin, IDBin], + NameAndID = [NameBin, IDType], [ cons([NodeID, <<"ets_table">>, <<"size">> | NameAndID], Size , Timestamp) , cons([NodeID, <<"ets_table">>, <<"memory">> | NameAndID], Memory, Timestamp) ]. @@ -167,6 +179,7 @@ of_processes( OfProcess = fun (P) -> of_process(P, NodeID, Timestamp) end, PerProcessMsgsNested = lists:map(OfProcess, Processes), PerProcessMsgsFlattened = lists:append(PerProcessMsgsNested), + PerProcessMsgsAggregates = aggregate_by_path(PerProcessMsgsFlattened, Timestamp), Ts = Timestamp, N = NodeID, [ cons([N, <<"processes_count_all">> ], CountAll , Ts) @@ -177,38 +190,50 @@ of_processes( , cons([N, <<"processes_count_running">> ], CountRunning , Ts) , cons([N, <<"processes_count_suspended">> ], CountSuspended , Ts) , cons([N, <<"processes_count_waiting">> ], CountWaiting , Ts) - | PerProcessMsgsFlattened + | PerProcessMsgsAggregates ]. -spec of_process(beam_stats_process:t(), binary(), erlang:timestamp()) -> [t()]. of_process( #beam_stats_process - { pid = Pid + { pid = _ , memory = Memory , total_heap_size = TotalHeapSize , stack_size = StackSize , message_queue_len = MsgQueueLen + , reductions = Reductions }=Process, <>, Timestamp ) -> Origin = beam_stats_process:get_best_known_origin(Process), OriginBin = proc_origin_to_bin(Origin), - PidBin = pid_to_bin(Pid), - OriginAndPid = [OriginBin, PidBin], Ts = Timestamp, N = NodeID, - [ cons([N, <<"process_memory">> | OriginAndPid], Memory , Ts) - , cons([N, <<"process_total_heap_size">> | OriginAndPid], TotalHeapSize , Ts) - , cons([N, <<"process_stack_size">> | OriginAndPid], StackSize , Ts) - , cons([N, <<"process_message_queue_len">> | OriginAndPid], MsgQueueLen , Ts) + [ cons([N, <<"process_memory">> , OriginBin], Memory , Ts) + , cons([N, <<"process_total_heap_size">> , OriginBin], TotalHeapSize , Ts) + , cons([N, <<"process_stack_size">> , OriginBin], StackSize , Ts) + , cons([N, <<"process_message_queue_len">> , OriginBin], MsgQueueLen , Ts) + , cons([N, <<"process_reductions">> , OriginBin], Reductions , Ts) ]. +-spec aggregate_by_path([t()], erlang:timestamp()) -> + [t()]. +aggregate_by_path(Msgs, Timestamp) -> + Aggregate = + fun (?T{path=K, value=V}, ValsByPath) -> + dict:update_counter(K, V, ValsByPath) + end, + ValsByPathDict = lists:foldl(Aggregate, dict:new(), Msgs), + ValsByPathList = dict:to_list(ValsByPathDict), + [cons(Path, Value, Timestamp) || {Path, Value} <- ValsByPathList]. + -spec proc_origin_to_bin(beam_stats_process:best_known_origin()) -> binary(). proc_origin_to_bin({registered_name, Name}) -> - atom_to_binary(Name, utf8); + NameBin = atom_to_binary(Name, utf8), + <<"named--", NameBin/binary>>; proc_origin_to_bin({ancestry, Ancestry}) -> #beam_stats_process_ancestry { raw_initial_call = InitCallRaw @@ -221,7 +246,8 @@ proc_origin_to_bin({ancestry, Ancestry}) -> AncestorsBinOpt = hope_option:map(AncestorsOpt , fun ancestors_to_bin/1), AncestorsBin = hope_option:get(AncestorsBinOpt , Blank), InitCallRawBin = mfa_to_bin(InitCallRaw), - << InitCallRawBin/binary + << "spawned-via--" + , InitCallRawBin/binary , "--" , InitCallOTPBin/binary , "--" @@ -242,12 +268,7 @@ ancestors_to_bin([A | Ancestors]) -> ancestor_to_bin(A) when is_atom(A) -> atom_to_binary(A, utf8); ancestor_to_bin(A) when is_pid(A) -> - pid_to_bin(A). - -pid_to_bin(Pid) -> - PidList = erlang:pid_to_list(Pid), - PidBin = re:replace(PidList, "[\.]", "_", [global, {return, binary}]), - re:replace(PidBin , "[><]", "" , [global, {return, binary}]). + <<"PID">>. -spec mfa_to_bin(mfa()) -> binary().