Current section
Files
Jump to
Current section
Files
src/observer_cli_trace.erl
-module(observer_cli_trace).
-export([call/2, stop_all/0, wait_cleanup/1]).
-ifdef(TEST).
-export([
test_helpers/0,
clear_trace/0,
forced_reason/1,
outcome_events/1,
io_reply/1,
module_md5/1,
parse_pid/1,
format_event/6,
final_drain/1,
wait_formatter/3,
checked_helper_shutdown/3,
drain_forced/2,
outcome_payload/1,
owner_result/3,
finalize_failure_result/4,
owner_init/0,
verify_cleanup/1,
wait_fixed_names/0,
trace_pattern_off/1,
stop_helper_checked/2,
stop_helper/1,
await_timeout_cleanup/6,
await_owner_down/4,
fallback_cleanup/3,
wait_trace/3,
stop_owner/2,
await_stop_ack/6,
await_helper_down/2,
await_killed_helper/2,
monitor_failure/3,
wait_fixed_names/1,
silent_io/0
]).
-endif.
-define(OWNER, observer_cli_trace_owner).
-define(RECON_VERSION, "2.5.6").
-define(MAX_EVENTS, 1000).
-define(ACK_TIMEOUT_MS, 1000).
-define(STOP_TIMEOUT_MS, 2000).
-define(CLEAR_RESULT_KEY, observer_cli_trace_clear_result).
-spec call(pid(), map()) -> map().
call(Controller, Request) when is_pid(Controller), is_map(Request) ->
case validate(Request) of
{ok, Trace} -> admit(Controller, Trace);
{error, Category, Reason} -> response(error, Category, Reason, null, [])
end;
call(_Controller, _Request) ->
response(error, argument, invalid_request, null, []).
-spec stop_all() -> map().
stop_all() ->
Warning = global_warning(),
case recon_capability() of
ok -> do_stop_all(Warning);
{error, Reason} -> response(error, capability, Reason, null, [Warning])
end.
do_stop_all(Warning) ->
case whereis(?OWNER) of
undefined ->
clear_trace(),
_ = wait_fixed_names(),
response(
error,
cleanup,
cleanup_unconfirmed,
(forced_capture(cleanup_unconfirmed, []))#{cleanup_confirmed => false},
[Warning]
);
Owner ->
stop_owner(Owner, Warning)
end.
validate(Request) ->
case maps:get(replace_existing_trace, Request, false) of
true ->
case recon_capability(Request) of
ok -> validate_mfa(Request);
{error, Reason} -> {error, capability, Reason}
end;
false ->
{error, argument, replace_existing_trace_required}
end.
recon_capability() ->
recon_capability(#{}).
recon_capability(Request) ->
_ = application:load(recon),
ExpectedVersion = expected_recon_version(Request),
case
{
application:get_key(recon, vsn),
code:ensure_loaded(recon_trace)
}
of
{{ok, ExpectedVersion}, {module, recon_trace}} ->
case
erlang:function_exported(recon_trace, calls, 3) andalso
erlang:function_exported(recon_trace, clear, 0)
of
true -> ok;
false -> {error, capability_unavailable}
end;
_ ->
{error, capability_unavailable}
end.
validate_mfa(#{mfa := Text} = Request) ->
case parse_mfa(Text) of
{ok, {Module, Function, Arity} = MFA} ->
case
is_tuple(code:is_loaded(Module)) andalso
erlang:function_exported(Module, Function, Arity)
of
true -> validate_pid(Request, MFA);
false -> {error, capability, mfa_unavailable}
end;
error ->
{error, argument, invalid_mfa}
end;
validate_mfa(_Request) ->
{error, argument, invalid_mfa}.
parse_mfa(Text) when is_binary(Text) ->
parse_mfa(binary_to_list(Text));
parse_mfa(Text) when is_list(Text), length(Text) =< 768 ->
case string:split(Text, ":", all) of
[ModuleText, FunctionArity] ->
case string:split(FunctionArity, "/", all) of
[FunctionText, ArityText] -> existing_mfa(ModuleText, FunctionText, ArityText);
_ -> error
end;
_ ->
error
end;
parse_mfa(_Text) ->
error.
existing_mfa(ModuleText, FunctionText, ArityText) ->
try
true = exact_name(ModuleText),
true = exact_name(FunctionText),
Module = list_to_existing_atom(ModuleText),
Function = list_to_existing_atom(FunctionText),
Arity = list_to_integer(ArityText),
true = Arity >= 0 andalso Arity =< 255,
{ok, {Module, Function, Arity}}
catch
_:_ -> error
end.
exact_name(Name) ->
Name =/= [] andalso Name =/= "_" andalso Name =/= "*".
validate_pid(#{pid := Text} = Request, MFA) ->
case parse_pid(Text) of
{ok, Pid} -> validate_bounds(Request, MFA, Pid);
error -> {error, safety_refusal, invalid_trace_pid}
end;
validate_pid(_Request, _MFA) ->
{error, argument, trace_pid_required}.
parse_pid(Text) when is_binary(Text) ->
parse_pid(binary_to_list(Text));
parse_pid(Text) when is_list(Text), length(Text) =< 128 ->
try list_to_pid(Text) of
Pid when node(Pid) =:= node() ->
case is_process_alive(Pid) of
true -> {ok, Pid};
false -> error
end;
_ ->
error
catch
_:_ -> error
end;
parse_pid(_Text) ->
error.
validate_bounds(Request, MFA, Pid) ->
Duration = maps:get(duration_ms, Request, 10000),
Max = maps:get(max, Request, 100),
case valid_duration(Duration) andalso valid_max(Max) of
true ->
Trace = #{
mfa => MFA,
pid => Pid,
duration_ms => Duration,
max => Max,
started_timestamp => os:timestamp(),
module_md5 => module_md5(element(1, MFA))
},
{ok, test_request_options(Request, Trace)};
false ->
{error, argument, invalid_trace_bounds}
end.
valid_duration(Duration) ->
is_integer(Duration) andalso Duration >= 100 andalso Duration =< 60000.
valid_max(Max) when is_integer(Max) -> Max >= 1 andalso Max =< 1000;
valid_max({Count, 1000}) -> is_integer(Count) andalso Count >= 1 andalso Count =< 200;
valid_max(_Max) -> false.
module_md5(Module) ->
try erlang:get_module_info(Module, md5) of
Md5 when is_binary(Md5) -> Md5
catch
error:badarg -> null
end.
admit(Controller, Trace) ->
Dispatcher = self(),
Owner = spawn(fun owner_init/0),
try register(?OWNER, Owner) of
true ->
Ref = make_ref(),
OwnerRef = erlang:monitor(process, Owner),
Owner ! {start, Dispatcher, Controller, Ref, Trace},
dispatch_loop(Owner, OwnerRef, Ref, Trace, [], false)
catch
error:badarg ->
exit(Owner, kill),
response(error, safety_refusal, trace_busy, null, [global_warning()])
end.
dispatch_loop(Owner, OwnerRef, Ref, Trace, Helpers, Entered) ->
receive
{Ref, ready, Manifest} ->
dispatch_loop(Owner, OwnerRef, Ref, Trace, maps:values(Manifest), Entered);
{Ref, calls_entered} ->
dispatch_loop(Owner, OwnerRef, Ref, Trace, Helpers, true);
{Ref, result, Result} ->
await_owner_down(Owner, OwnerRef, Ref, Result);
{'DOWN', OwnerRef, process, Owner, _Reason} ->
fallback_cleanup(Trace, Helpers, Entered)
after dispatch_timeout(Trace) ->
Owner ! {force_stop, Ref, dispatcher_timeout},
await_timeout_cleanup(Owner, OwnerRef, Ref, Trace, Helpers, Entered)
end.
await_timeout_cleanup(Owner, OwnerRef, Ref, Trace, Helpers, Entered) ->
receive
{Ref, result, Result} ->
await_owner_down(Owner, OwnerRef, Ref, Result);
{'DOWN', OwnerRef, process, Owner, _Reason} ->
fallback_cleanup(Trace, Helpers, Entered)
after ?STOP_TIMEOUT_MS ->
response(
error,
cleanup,
cleanup_unconfirmed,
(with_selector(Trace, forced_capture(cleanup_unconfirmed, [])))#{
cleanup_confirmed => false
},
[global_warning()]
)
end.
await_owner_down(Owner, OwnerRef, Ref, Result) ->
receive
{'DOWN', OwnerRef, process, Owner, _Reason} ->
case whereis(?OWNER) of
undefined -> Result;
_ -> response(error, cleanup, cleanup_unconfirmed, null, [global_warning()])
end;
{Ref, ready, _Manifest} ->
await_owner_down(Owner, OwnerRef, Ref, Result)
after ?STOP_TIMEOUT_MS ->
response(error, cleanup, cleanup_unconfirmed, null, [global_warning()])
end.
fallback_cleanup(Trace, Helpers, Entered) ->
case Entered of
true -> clear_trace();
false -> ok
end,
stop_helpers(Helpers),
case Entered andalso verify_cleanup(Trace) =:= ok of
true ->
Capture = with_selector(Trace, forced_capture(capture_internal_error, [])),
response(error, internal, capture_internal_error, Capture, [global_warning()]);
false ->
Capture = (with_selector(Trace, forced_capture(cleanup_unconfirmed, [])))#{
cleanup_confirmed => false
},
response(error, cleanup, cleanup_unconfirmed, Capture, [global_warning()])
end.
owner_init() ->
receive
{start, Dispatcher, Controller, Ref, Trace} ->
owner_start(Dispatcher, Controller, Ref, Trace)
after ?ACK_TIMEOUT_MS ->
ok
end.
owner_start(Dispatcher, Controller, Ref, Trace) ->
process_flag(trap_exit, true),
DispatcherMon = erlang:monitor(process, Dispatcher),
ControllerMon = erlang:monitor(process, Controller),
TraceeMon = erlang:monitor(process, maps:get(pid, Trace)),
Session = make_ref(),
CollectorCap = maps:get(test_event_cap, Trace, ?MAX_EVENTS),
Collector = spawn_link(fun() -> collector([], 0, 0, false, CollectorCap) end),
SilentIO = spawn_link(fun silent_io/0),
try
CollectorMon = erlang:monitor(process, Collector),
SilentMon = erlang:monitor(process, SilentIO),
true = group_leader(SilentIO, self()),
Manifest = #{collector => Collector, silent_io => SilentIO},
put(observer_cli_trace_helpers, Manifest),
State = Trace#{
dispatcher => Dispatcher,
dispatcher_mon => DispatcherMon,
controller_mon => ControllerMon,
tracee_mon => TraceeMon,
collector => Collector,
collector_mon => CollectorMon,
silent_io => SilentIO,
silent_mon => SilentMon,
session => Session,
ref => Ref
},
test_before_ready(State),
Dispatcher ! {Ref, ready, Manifest},
Result = run_owner(State),
Dispatcher ! {Ref, result, Result}
after
stop_helpers([Collector, SilentIO])
end.
run_owner(State) ->
Dispatcher = maps:get(dispatcher, State),
Ref = maps:get(ref, State),
case prepare_owner(State) of
ok ->
Dispatcher ! {Ref, calls_entered},
Outcome = entered_trace(State),
finish_owner_safely(State, Outcome);
error ->
stop_helpers([maps:get(collector, State), maps:get(silent_io, State)]),
response(error, internal, helper_setup_failed, null, [global_warning()])
end.
prepare_owner(State) ->
try test_before_calls(State) of
_ -> ok
catch
_:_ -> error
end.
entered_trace(State) ->
Outcome =
try
Matches = start_recon(State),
test_after_calls(State),
case Matches of
0 ->
{error, capability, mfa_not_traceable};
_ ->
Recon = recon_processes(),
case {maps:get(tracer, Recon), maps:get(formatter, Recon)} of
{Tracer, Formatter} when is_pid(Tracer), is_pid(Formatter) ->
wait_trace(State, Recon, deadline(State));
{undefined, Formatter} when is_pid(Formatter) ->
wait_formatter(State, Recon, deadline(State));
{_Tracer, undefined} ->
completed_before_attach(State);
_ ->
setup_failure_outcome(State)
end
end
catch
_:_ -> setup_failure_outcome(State)
after
clear_trace()
end,
drain_forced(State, Outcome).
setup_failure_outcome(State) ->
receive
{'DOWN', Mon, process, _Pid, Reason} -> monitor_failure(State, Mon, Reason)
after 0 ->
{forced, internal, capture_internal_error}
end.
finish_owner_safely(State, Outcome) ->
try finish_owner(State, Outcome) of
Result -> Result
catch
_:_ -> finalize_failure(State, Outcome)
end.
finalize_failure(State, Outcome) ->
clear_trace(),
stop_helpers([maps:get(collector, State), maps:get(silent_io, State)]),
{Events, Rejected} = outcome_payload(Outcome),
finalize_failure_result(State, Events, Rejected, verify_cleanup(State)).
finalize_failure_result(State, Events, Rejected, Verification) ->
case Verification of
ok ->
Capture = with_selector(
State, forced_capture(capture_internal_error, Events, Rejected)
),
response(error, internal, capture_internal_error, Capture, [global_warning()]);
{error, _Reason} ->
Capture = (with_selector(
State, forced_capture(cleanup_unconfirmed, Events, Rejected)
))#{
cleanup_confirmed => false
},
response(
error,
cleanup,
cleanup_unconfirmed,
Capture,
[global_warning()]
)
end.
start_recon(State) ->
Collector = maps:get(collector, State),
Session = maps:get(session, State),
Started = maps:get(started_timestamp, State),
ExpectedPid = maps:get(pid, State),
ExpectedMFA = maps:get(mfa, State),
Formatter = fun(Message) ->
format_event(Message, ExpectedPid, ExpectedMFA, Collector, Session, Started)
end,
ok = clear_fixed_name_ports(),
clear_extended_patterns(),
recon_trace:calls(
ExpectedMFA,
maps:get(max, State),
[
{pid, ExpectedPid},
{args, arity},
{timestamp, trace},
{scope, global},
{formatter, Formatter},
{io_server, maps:get(silent_io, State)}
]
).
recon_processes() ->
Tracer = whereis(recon_trace_tracer),
Formatter = whereis(recon_trace_formatter),
#{
tracer => Tracer,
tracer_mon => monitor_if_pid(Tracer),
formatter => Formatter,
formatter_mon => monitor_if_pid(Formatter)
}.
monitor_if_pid(Pid) when is_pid(Pid) -> erlang:monitor(process, Pid);
monitor_if_pid(_Pid) -> undefined.
deadline(State) ->
erlang:monotonic_time(millisecond) + maps:get(duration_ms, State).
wait_trace(State, Recon, Deadline) ->
Tracer = maps:get(tracer, Recon),
TracerMon = maps:get(tracer_mon, Recon),
Ref = maps:get(ref, State),
receive
{'DOWN', TracerMon, process, Tracer, normal} when is_reference(TracerMon) ->
wait_formatter(State, Recon#{tracer_exit_reason => normal}, Deadline);
{'DOWN', TracerMon, process, Tracer, noproc} when is_reference(TracerMon) ->
wait_formatter(State, Recon, Deadline);
{'DOWN', TracerMon, process, Tracer, _Reason} when is_reference(TracerMon) ->
{forced, internal, capture_internal_error};
{'EXIT', Tracer, Reason} when is_pid(Tracer) ->
wait_trace(State, Recon#{tracer_exit_reason => Reason}, Deadline);
{stop_request, Stopper, StopRef} ->
Stopper ! {StopRef, armed, self()},
{forced, success, stopped, {Stopper, StopRef}};
{force_stop, Ref, Reason} ->
forced_reason(Reason);
{'DOWN', Mon, process, _Pid, Reason} ->
monitor_failure(State, Mon, Reason)
after remaining(Deadline) ->
{forced, success, duration_elapsed}
end.
wait_formatter(State, Recon, Deadline) ->
Formatter = maps:get(formatter, Recon),
FormatterMon = maps:get(formatter_mon, Recon),
StateRef = maps:get(ref, State),
case FormatterMon of
undefined ->
completed_before_attach(State, maps:get(tracer_exit_reason, Recon, unknown));
_ ->
receive
{'DOWN', FormatterMon, process, Formatter, normal} ->
completed_before_attach(State, normal);
{'DOWN', FormatterMon, process, Formatter, noproc} ->
completed_before_attach(
State, maps:get(tracer_exit_reason, Recon, unknown)
);
{'DOWN', FormatterMon, process, Formatter, _Reason} ->
{forced, internal, capture_internal_error};
{stop_request, Stopper, StopRef} ->
Stopper ! {StopRef, armed, self()},
{forced, success, stopped, {Stopper, StopRef}};
{force_stop, StateRef, Reason} ->
forced_reason(Reason);
{'DOWN', Mon, process, _Pid, Reason} ->
monitor_failure(State, Mon, Reason)
after remaining(Deadline) ->
{forced, success, duration_elapsed}
end
end.
final_drain(State) ->
Collector = maps:get(collector, State),
CollectorMon = maps:get(collector_mon, State),
Ref = make_ref(),
Collector ! {final, self(), Ref},
receive
{Ref, Events, Count, Rejected, Truncated} ->
{natural, natural_reason(maps:get(max, State)), Events, Truncated,
Count - length(Events), Rejected};
{'DOWN', CollectorMon, process, Collector, _Reason} ->
{forced, internal, capture_internal_error}
after ?ACK_TIMEOUT_MS ->
{forced, internal, capture_internal_error}
end.
completed_before_attach(State) ->
completed_before_attach(State, unknown).
completed_before_attach(State, ExitReason) ->
case final_drain(State) of
{natural, _Reason, Events, _Truncated, Dropped, Rejected} = Natural ->
Forwarded = length(Events) + Dropped + Rejected,
case
natural_completion_confirmed(
maps:get(max, State), Forwarded, State, ExitReason
)
of
true -> Natural;
false -> with_forced_payload(setup_failure_outcome(State), Events, Rejected)
end;
Failure ->
Failure
end.
natural_threshold_reached(Max, Forwarded) when is_integer(Max) ->
Forwarded >= Max;
natural_threshold_reached({Max, _Window}, Forwarded) ->
Forwarded >= Max + 1.
natural_completion_confirmed(Max, Forwarded, _State, _ExitReason) when is_integer(Max) ->
natural_threshold_reached(Max, Forwarded);
natural_completion_confirmed(Max, Forwarded, State, ExitReason) ->
natural_threshold_reached(Max, Forwarded) andalso
recon_exit_reason(State, ExitReason) =:= normal.
recon_exit_reason(_State, Reason) when Reason =/= unknown ->
Reason;
recon_exit_reason(State, unknown) ->
Collector = maps:get(collector, State),
SilentIO = maps:get(silent_io, State, undefined),
receive
{'EXIT', Pid, Reason} when Pid =/= Collector, Pid =/= SilentIO -> Reason
after ?ACK_TIMEOUT_MS ->
unknown
end.
with_forced_payload({forced, Category, Reason}, Events, Rejected) ->
{forced, Category, Reason, Events, Rejected}.
drain_forced(State, {forced, Category, Reason}) ->
case final_drain(State) of
{natural, _NaturalReason, Events, _Truncated, _Dropped, Rejected} ->
{forced, Category, Reason, Events, Rejected};
Failure ->
Failure
end;
drain_forced(State, {forced, Category, Reason, Stopper}) ->
case final_drain(State) of
{natural, _NaturalReason, Events, _Truncated, _Dropped, Rejected} ->
{forced, Category, Reason, Events, Rejected, Stopper};
Failure ->
Failure
end;
drain_forced(_State, Outcome) ->
Outcome.
natural_reason(Max) when is_integer(Max) -> limit_reached;
natural_reason({_Count, 1000}) -> rate_exceeded.
forced_reason(controller_disconnected) -> {forced, controller, controller_disconnected};
forced_reason(dispatcher_timeout) -> {forced, internal, capture_internal_error};
forced_reason(_Reason) -> {forced, internal, capture_internal_error}.
monitor_failure(State, Mon, _Reason) ->
ControllerMon = maps:get(controller_mon, State),
DispatcherMon = maps:get(dispatcher_mon, State),
TraceeMon = maps:get(tracee_mon, State),
case Mon of
ControllerMon ->
{forced, controller, controller_disconnected};
DispatcherMon ->
{forced, internal, dispatcher_disconnected};
TraceeMon ->
{forced, safety_refusal, tracee_exited};
_ ->
{forced, internal, capture_internal_error}
end.
finish_owner(State, Outcome) ->
Verification = verify_cleanup(State),
test_before_helper_stop(State),
{CheckedOutcome, CheckedVerification} = checked_helper_shutdown(
State, Outcome, Verification
),
Result = owner_result(State, CheckedOutcome, CheckedVerification),
notify_stopper(Outcome, Result),
Result.
checked_helper_shutdown(State, Outcome, Verification) ->
Collector = stop_helper_checked(maps:get(collector, State), maps:get(collector_mon, State)),
SilentIO = stop_helper_checked(maps:get(silent_io, State), maps:get(silent_mon, State)),
case {Verification, Collector, SilentIO} of
{ok, ok, ok} -> {Outcome, ok};
{{error, _Reason}, _, _} -> {Outcome, {error, cleanup_unconfirmed}};
{_, cleanup_unconfirmed, _} -> {Outcome, {error, cleanup_unconfirmed}};
{_, _, cleanup_unconfirmed} -> {Outcome, {error, cleanup_unconfirmed}};
{ok, helper_failed, _} -> {helper_failure_outcome(Outcome), ok};
{ok, _, helper_failed} -> {helper_failure_outcome(Outcome), ok}
end.
helper_failure_outcome(Outcome) ->
{Events, Rejected} = outcome_payload(Outcome),
{forced, internal, capture_internal_error, Events, Rejected}.
outcome_payload({natural, _Reason, Events, _Truncated, _Dropped, Rejected}) ->
{Events, Rejected};
outcome_payload({forced, _Category, _Reason, Events, Rejected}) ->
{Events, Rejected};
outcome_payload({forced, _Category, _Reason, Events, Rejected, _Stopper}) ->
{Events, Rejected};
outcome_payload(_Outcome) ->
{[], 0}.
owner_result(State, Outcome, ok) ->
EndMd5 = end_module_md5(State),
Partial = EndMd5 =/= maps:get(module_md5, State),
case Outcome of
{natural, Reason, Events, Truncated, Dropped, Rejected} ->
Incomplete = Partial orelse Truncated orelse Dropped > 0 orelse Rejected > 0,
Capture = #{
status =>
case Incomplete of
true -> partial;
false -> complete
end,
reason => Reason,
trace_complete => not Incomplete,
truncated => Truncated orelse Rejected > 0,
dropped_count =>
case Rejected of
0 -> Dropped;
_ -> null
end,
events => Events,
module_reloaded => Partial,
interference_detected => Rejected > 0,
coverage => external_global_calls_only,
cleanup_confirmed => true
},
response(ok, success, Reason, with_selector(State, Capture), [global_warning()]);
{forced, Category, Reason, Events, Rejected} ->
response(
forced_status(Category),
Category,
Reason,
with_selector(State, forced_capture(Reason, Events, Rejected)),
[global_warning()]
);
{forced, Category, Reason, Events, Rejected, _Stopper} ->
response(
forced_status(Category),
Category,
Reason,
with_selector(State, forced_capture(Reason, Events, Rejected)),
[global_warning()]
);
{error, Category, Reason} ->
response(error, Category, Reason, null, [global_warning()])
end;
owner_result(State, Outcome, {error, _Reason}) ->
{Events, Rejected} = outcome_payload(Outcome),
Capture = (with_selector(State, forced_capture(cleanup_unconfirmed, Events, Rejected)))#{
cleanup_confirmed => false
},
response(error, cleanup, cleanup_unconfirmed, Capture, [global_warning()]).
forced_status(success) -> ok;
forced_status(_Category) -> error.
-ifdef(TEST).
outcome_events({natural, _Reason, Events, _Truncated, _Dropped, _Rejected}) -> Events;
outcome_events({forced, _Category, _Reason, Events, _Rejected}) -> Events;
outcome_events({forced, _Category, _Reason, Events, _Rejected, _Stopper}) -> Events;
outcome_events(_Outcome) -> [].
-endif.
forced_capture(Reason, Events) ->
forced_capture(Reason, Events, 0).
forced_capture(Reason, Events, Rejected) ->
#{
status => partial,
reason => Reason,
trace_complete => false,
truncated => true,
dropped_count => null,
events => Events,
interference_detected => Rejected > 0,
coverage => external_global_calls_only,
cleanup_confirmed => true
}.
with_selector(#{mfa := {Module, Function, Arity}, pid := Pid}, Capture) ->
Capture#{
tracee => {identifier, pid, Pid},
mfa => {mfa, Module, Function, Arity}
};
with_selector(_State, Capture) ->
Capture.
notify_stopper({forced, _Category, _Reason, _Events, _Rejected, {Stopper, StopRef}}, Result) ->
Stopper ! {StopRef, cleanup_ack, Result};
notify_stopper(_Outcome, _Result) ->
ok.
verify_cleanup(State) ->
case get(?CLEAR_RESULT_KEY) of
{error, _Reason} = Error ->
Error;
_ ->
case wait_fixed_names() of
ok ->
Pid = maps:get(pid, State),
MFA = maps:get(mfa, State),
case {call_flag_off(Pid), trace_pattern_off(MFA)} of
{true, true} -> ok;
_ -> {error, cleanup_unconfirmed}
end;
Error ->
Error
end
end.
wait_fixed_names() ->
wait_fixed_names(100).
wait_fixed_names(0) ->
{error, cleanup_unconfirmed};
wait_fixed_names(Attempts) ->
case {whereis(recon_trace_tracer), whereis(recon_trace_formatter)} of
{undefined, undefined} ->
ok;
_ ->
timer:sleep(10),
wait_fixed_names(Attempts - 1)
end.
call_flag_off(Pid) ->
case erlang:trace_info(Pid, flags) of
undefined -> true;
{flags, Flags} -> not lists:member(call, Flags)
end.
trace_pattern_off(MFA) ->
case erlang:trace_info(MFA, traced) of
undefined -> true;
{traced, false} -> true;
{traced, undefined} -> true;
_ -> false
end.
format_event(
{trace_ts, Tracee, call, {Module, Function, Arity} = MFA, Timestamp},
Tracee,
MFA,
Collector,
Session,
Started
) when
is_pid(Tracee), is_atom(Module), is_atom(Function), is_integer(Arity)
->
EventRef = make_ref(),
Offset = max(0, timer:now_diff(Timestamp, Started) div 1000),
Event = #{
tracee => {identifier, pid, Tracee},
mfa => {mfa, Module, Function, Arity},
offset_ms => Offset
},
Collector ! {event, self(), Session, EventRef, Event},
await_collector(Collector, Session, EventRef);
format_event(_Message, _ExpectedPid, _ExpectedMFA, Collector, Session, _Started) ->
EventRef = make_ref(),
Collector ! {interference, self(), Session, EventRef},
await_collector(Collector, Session, EventRef).
await_collector(Collector, Session, EventRef) ->
receive
{Session, EventRef, ack} -> [];
{'DOWN', _Mon, process, Collector, _Reason} -> erlang:error(collector_down)
after ?ACK_TIMEOUT_MS ->
erlang:error(collector_ack_timeout)
end.
collector(Events, Count, Rejected, Truncated, Cap) ->
receive
{event, Formatter, Session, EventRef, Event} ->
{NextEvents, NextTruncated} =
case Count < Cap of
true -> {[Event | Events], Truncated};
false -> {Events, true}
end,
Formatter ! {Session, EventRef, ack},
collector(NextEvents, Count + 1, Rejected, NextTruncated, Cap);
{interference, Formatter, Session, EventRef} ->
Formatter ! {Session, EventRef, ack},
collector(Events, Count, Rejected + 1, Truncated, Cap);
{final, Owner, Ref} ->
Owner ! {Ref, lists:reverse(Events), Count, Rejected, Truncated},
collector(Events, Count, Rejected, Truncated, Cap);
{test_count, From, Ref} ->
From ! {Ref, Count},
collector(Events, Count, Rejected, Truncated, Cap);
{stop, Owner, Ref} ->
Owner ! {Ref, stopping},
ok;
stop ->
ok
end.
silent_io() ->
receive
{io_request, From, ReplyAs, Request} ->
From ! {io_reply, ReplyAs, io_reply(Request)},
silent_io();
{stop, Owner, Ref} ->
Owner ! {Ref, stopping},
ok;
stop ->
ok;
_Other ->
silent_io()
end.
io_reply({put_chars, _Characters}) -> ok;
io_reply({put_chars, _Encoding, _Characters}) -> ok;
io_reply({put_chars, Module, Function, Args}) -> io_apply(Module, Function, Args);
io_reply({put_chars, _Encoding, Module, Function, Args}) -> io_apply(Module, Function, Args);
io_reply({requests, Requests}) -> io_requests(Requests);
io_reply({get_geometry, _}) -> {error, enotsup};
io_reply(_Request) -> {error, enotsup}.
io_apply(Module, Function, Args) ->
try erlang:apply(Module, Function, Args) of
_Characters -> ok
catch
_:_ -> {error, request}
end.
io_requests([]) ->
ok;
io_requests([Request | Rest]) ->
case io_reply(Request) of
ok -> io_requests(Rest);
Error -> Error
end.
stop_helpers(Pids) ->
lists:foreach(fun stop_helper/1, Pids).
stop_helper_checked(Pid, Mon) ->
case is_process_alive(Pid) of
false ->
erlang:demonitor(Mon, [flush]),
helper_failed;
true ->
Ref = make_ref(),
Pid ! {stop, self(), Ref},
receive
{Ref, stopping} -> await_helper_down(Pid, Mon);
{'DOWN', Mon, process, Pid, _Reason} -> helper_failed
after 50 ->
exit(Pid, kill),
await_killed_helper(Pid, Mon)
end
end.
await_helper_down(Pid, Mon) ->
receive
{'DOWN', Mon, process, Pid, normal} -> ok;
{'DOWN', Mon, process, Pid, _Reason} -> helper_failed
after ?ACK_TIMEOUT_MS ->
exit(Pid, kill),
await_killed_helper(Pid, Mon)
end.
await_killed_helper(Pid, Mon) ->
receive
{'DOWN', Mon, process, Pid, _Reason} -> cleanup_unconfirmed
after ?ACK_TIMEOUT_MS ->
erlang:demonitor(Mon, [flush]),
cleanup_unconfirmed
end.
stop_helper(Pid) when is_pid(Pid) ->
Mon = erlang:monitor(process, Pid),
Pid ! stop,
receive
{'DOWN', Mon, process, Pid, _Reason} -> ok
after 50 ->
exit(Pid, kill),
receive
{'DOWN', Mon, process, Pid, _Reason} -> ok
after ?ACK_TIMEOUT_MS ->
erlang:demonitor(Mon, [flush])
end
end;
stop_helper(_Other) ->
ok.
stop_owner(Owner, Warning) ->
OwnerMon = erlang:monitor(process, Owner),
StopRef = make_ref(),
Owner ! {stop_request, self(), StopRef},
receive
{StopRef, armed, Owner} ->
clear_trace(),
await_stop_ack(Owner, OwnerMon, StopRef, Warning, undefined, false);
{'DOWN', OwnerMon, process, Owner, _Reason} ->
clear_trace(),
_ = wait_fixed_names(),
response(
error,
cleanup,
cleanup_unconfirmed,
(forced_capture(cleanup_unconfirmed, []))#{cleanup_confirmed => false},
[Warning]
)
after 500 ->
clear_trace(),
_ = wait_fixed_names(),
erlang:demonitor(OwnerMon, [flush]),
response(
error,
cleanup,
cleanup_unconfirmed,
(forced_capture(cleanup_unconfirmed, []))#{cleanup_confirmed => false},
[Warning]
)
end.
await_stop_ack(Owner, OwnerMon, StopRef, Warning, Result, Down) ->
case {Result, Down} of
{Response, true} when is_map(Response) -> Response;
_ ->
receive
{StopRef, cleanup_ack, Response} ->
await_stop_ack(Owner, OwnerMon, StopRef, Warning, Response, Down);
{'DOWN', OwnerMon, process, Owner, _Reason} ->
await_stop_ack(Owner, OwnerMon, StopRef, Warning, Result, true)
after ?STOP_TIMEOUT_MS ->
response(
error,
cleanup,
cleanup_unconfirmed,
(forced_capture(cleanup_unconfirmed, []))#{cleanup_confirmed => false},
[Warning]
)
end
end.
remaining(Deadline) ->
max(0, Deadline - erlang:monotonic_time(millisecond)).
dispatch_timeout(Trace) ->
maps:get(test_dispatch_timeout_ms, Trace, maps:get(duration_ms, Trace) + 5000).
response(Status, Category, Reason, Capture, Warnings) ->
#{
status => Status,
category => Category,
reason => Reason,
capture => Capture,
warnings => Warnings
}.
global_warning() ->
#{
code => global_trace_replacement,
message => <<
"This command clears legacy process/port trace flags and tracers plus static call ",
"patterns, including on-load and call-memory patterns; recon 2.5.6 may also kill ",
"processes or ports occupying its fixed tracer or formatter names. Dynamic trace ",
"sessions are not directly cleared, but killing such an occupant can disable one."
>>
}.
clear_trace() ->
PortResult = clear_fixed_name_ports(),
LegacyResult =
case PortResult of
ok ->
try
recon_trace:clear()
catch
_:_ -> {error, cleanup_unconfirmed}
end;
PortError ->
PortError
end,
ExtendedResult =
try clear_extended_patterns() of
ok -> ok
catch
_:_ -> {error, cleanup_unconfirmed}
end,
ClearResult =
case {LegacyResult, ExtendedResult} of
{ok, ok} -> ok;
_ -> {error, cleanup_unconfirmed}
end,
case ClearResult of
ok -> erase(?CLEAR_RESULT_KEY);
ClearError -> put(?CLEAR_RESULT_KEY, ClearError)
end,
ClearResult.
clear_fixed_name_ports() ->
case
{
clear_fixed_name_port(recon_trace_tracer),
clear_fixed_name_port(recon_trace_formatter)
}
of
{ok, ok} -> ok;
_ -> {error, cleanup_unconfirmed}
end.
clear_fixed_name_port(Name) ->
case whereis(Name) of
Port when is_port(Port) ->
_ =
try
erlang:port_close(Port)
catch
_:_ -> false
end,
wait_fixed_name_port(Name, Port, 100);
_ ->
ok
end.
wait_fixed_name_port(_Name, _Port, 0) ->
{error, cleanup_unconfirmed};
wait_fixed_name_port(Name, Port, Attempts) ->
case {whereis(Name), erlang:port_info(Port)} of
{Current, undefined} when Current =/= Port ->
ok;
_ ->
timer:sleep(10),
wait_fixed_name_port(Name, Port, Attempts - 1)
end.
clear_extended_patterns() ->
_ = erlang:trace_pattern(
on_load, false, [local, meta, call_count, call_time, call_memory]
),
_ = erlang:trace_pattern(on_load, false, []),
_ = erlang:trace_pattern({'_', '_', '_'}, false, [call_memory]),
ok.
-spec wait_cleanup(non_neg_integer()) -> boolean().
wait_cleanup(Timeout) when is_integer(Timeout), Timeout >= 0 ->
wait_cleanup_until(erlang:monotonic_time(millisecond) + Timeout).
wait_cleanup_until(Deadline) ->
case {whereis(?OWNER), whereis(recon_trace_tracer), whereis(recon_trace_formatter)} of
{undefined, undefined, undefined} ->
true;
_ ->
case erlang:monotonic_time(millisecond) < Deadline of
true ->
timer:sleep(10),
wait_cleanup_until(Deadline);
false ->
false
end
end.
-ifdef(TEST).
test_request_options(Request, Trace) ->
maps:merge(
Trace,
maps:with(
[
test_after_calls,
test_before_calls,
test_before_ready,
test_before_helper_stop,
test_dispatch_timeout_ms,
test_end_module_md5,
test_event_cap
],
Request
)
).
expected_recon_version(Request) ->
maps:get(test_recon_version, Request, ?RECON_VERSION).
end_module_md5(State) ->
maps:get(test_end_module_md5, State, module_md5(element(1, maps:get(mfa, State)))).
test_before_calls(State) ->
case maps:find(test_before_calls, State) of
{ok, Fun} -> Fun();
error -> ok
end.
test_after_calls(State) ->
case maps:find(test_after_calls, State) of
{ok, Fun} -> Fun();
error -> ok
end.
test_before_ready(State) ->
case maps:find(test_before_ready, State) of
{ok, Fun} -> Fun();
error -> ok
end.
test_before_helper_stop(State) ->
case maps:find(test_before_helper_stop, State) of
{ok, Fun} -> Fun();
error -> ok
end.
test_helpers() ->
case whereis(?OWNER) of
undefined ->
undefined;
Owner ->
case process_info(Owner, dictionary) of
{dictionary, Dictionary} ->
proplists:get_value(observer_cli_trace_helpers, Dictionary);
undefined ->
undefined
end
end.
-else.
test_request_options(_Request, Trace) ->
Trace.
test_before_calls(_State) ->
ok.
test_after_calls(_State) ->
ok.
test_before_ready(_State) ->
ok.
test_before_helper_stop(_State) ->
ok.
expected_recon_version(_Request) ->
?RECON_VERSION.
end_module_md5(State) ->
module_md5(element(1, maps:get(mfa, State))).
-endif.