-module(observer_cli_diagnostic). -export([capture/2]). -ifdef(TEST). -export([ build_report/4, limit_findings/1, reductions_context/2, scheduler_findings/1, application_trend/1, observation_report/6, valid_memory_sample/1, valid_application_sample/2, map_status/1, scheduler_status/1, scheduler_over_threshold/1, map_gauge_trend/3, entity_trends/4, rule_id/1, domain_name/1, recommendation/1, process_context/1, inventory/1, optional_status/1, distribution_status/1, optional_reason/2, sample_duration/1, pressure_window/2, same_generation/1, reductions_rates/2, rates/2, positive_step_ratio/1, metric_series_delta/3, first_field_reason/2, required_complete/1, valid_limit/1, skipped_checks/1, capture_observation_samples/6, capture_samples/3 ]). -endif. -define(DEFAULT_INTERVAL_MS, 1500). -define(RULESET, <<"observer_cli.quick">>). -define(RULESET_VERSION, 1). -define(CONTEXT_LIMIT, 20). -spec capture(map(), map()) -> map() | {probe_error, atom()}. capture(Request, #{controller := Controller} = Context) when is_map(Request) -> case maps:find(observe, Request) of {ok, Observe} -> capture_observation(Request, Context, Controller, Observe); error -> capture_quick(Request, Context, Controller) end; capture(_Request, _Context) -> {probe_error, invalid_request}. capture_quick(Request, Context, Controller) -> StartedAt = erlang:system_time(millisecond), Started = erlang:monotonic_time(millisecond), ModuleLoaded = is_tuple(code:is_loaded(?MODULE)), Interval = maps:get(interval_ms, Request, ?DEFAULT_INTERVAL_MS), Plan = [Started, Started + Interval], Samples = capture_samples(Request, Context, Plan), Finished = erlang:monotonic_time(millisecond), build_report( Samples, Plan, #{ started_at => rfc3339(StartedAt), finished_at => rfc3339(erlang:system_time(millisecond)), duration_ms => Finished - Started, controller => Controller, module_loaded_before_sample => ModuleLoaded }, distribution(Controller) ). capture_observation(Request, Context, Controller, Observe) -> case observation_duration(Observe) of {ok, Duration} -> Mode = observation_mode(Request), Count = case Mode of deep -> 7; _ -> 5 end, StartedAt = erlang:system_time(millisecond), Started = erlang:monotonic_time(millisecond), Plan = observation_plan(Started, Duration, Count), ModuleLoaded = is_tuple(code:is_loaded(?MODULE)), _ = observer_cli_snapshot:diagnostic_scheduler_flag(true), try Samples = capture_observation_samples(Request, Context, Plan), Holder = final_binary_holders(Mode, Request, Context), Finished = erlang:monotonic_time(millisecond), observation_report( Mode, Samples, Plan, Holder, #{ started_at => rfc3339(StartedAt), finished_at => rfc3339(erlang:system_time(millisecond)), duration_ms => Finished - Started, controller => Controller, module_loaded_before_sample => ModuleLoaded }, distribution(Controller) ) after _ = observer_cli_snapshot:diagnostic_scheduler_flag(false) end; error -> {probe_error, invalid_duration} end. observation_duration(Text) when is_list(Text) -> observation_duration(unicode:characters_to_binary(Text)); observation_duration(Text) when is_binary(Text) -> case re:run(Text, <<"^([0-9]+)(ms|s)?$">>, [{capture, [1, 2], binary}]) of {match, [Digits, <<"s">>]} -> valid_observation_ms(binary_to_integer(Digits) * 1000); {match, [Digits, _]} -> valid_observation_ms(binary_to_integer(Digits)); nomatch -> error end; observation_duration(_) -> error. valid_observation_ms(Duration) when Duration >= 5000, Duration =< 60000 -> {ok, Duration}; valid_observation_ms(_Duration) -> error. observation_mode(#{deep := true}) -> deep; observation_mode(#{app := _}) -> application; observation_mode(_) -> observation. observation_plan(Started, Duration, Count) -> [Started + ((Duration * Index) div (Count - 1)) || Index <- lists:seq(0, Count - 1)]. capture_observation_samples(Request, Context, Plan) -> capture_observation_samples(Request, Context, Plan, 0, undefined, []). capture_observation_samples(_Request, _Context, [], _Index, _PreviousFinish, Acc) -> lists:reverse(Acc); capture_observation_samples(Request, Context, [Target | Rest], Index, PreviousFinish, Acc) -> case PreviousFinish =/= undefined andalso PreviousFinish > Target of true -> Gap = #{status => error, reason_code => sampling_gap, target_monotonic_ms => Target}, capture_observation_samples(Request, Context, Rest, Index + 1, PreviousFinish, [ Gap | Acc ]); false -> sleep_until(Target), SchedulerEnd = case Acc of [] -> undefined; _ -> safe_scheduler_sample() end, Sample0 = capture_sample(Request, Context, Index, Target), SchedulerBaseline = safe_scheduler_sample(), Sample = Sample0#{ target_monotonic_ms => Target, scheduler_end => SchedulerEnd, scheduler_baseline => SchedulerBaseline }, Finish = maps:get(monotonic_finish_ms, Sample, erlang:monotonic_time(millisecond)), NextRequest = retain_application_refusal(Request, Sample), capture_observation_samples(NextRequest, Context, Rest, Index + 1, Finish, [ Sample | Acc ]) end. retain_application_refusal( #{app := _} = Request, #{application := #{status := unavailable, reason_code := scan_budget_exceeded} = Application} ) -> Request#{application_refusal => Application}; retain_application_refusal(Request, _Sample) -> Request. safe_scheduler_sample() -> try observer_cli_snapshot:diagnostic_scheduler_sample() catch _:_ -> #{status => error, reason_code => scheduler_sample_failed} end. final_binary_holders(deep, Request, Context) -> try observer_cli_snapshot:diagnostic_binary_holders(Request, Context) catch _:_ -> #{status => error, reason_code => binary_holder_scan_failed} end; final_binary_holders(_Mode, _Request, _Context) -> #{status => unavailable, reason_code => deep_not_requested}. capture_samples(Request, Context, [First, Second]) -> FirstSample = capture_sample(Request, Context, 0, First), case maps:get(monotonic_finish_ms, FirstSample, First) > Second of true -> [FirstSample, #{status => error, reason_code => sampling_gap}]; false -> [FirstSample, capture_sample(Request, Context, 1, Second)] end. capture_sample(Request, Context, Index, Target) -> sleep_until(Target), try sample_request(Request, Context, Index) of Sample -> Sample catch _Class:_Reason:_Stacktrace -> #{status => error, reason_code => required_probe_failed} end. sample_request(#{app := _, application_refusal := Application} = Request, Context, Index) -> Sample = sample(maps:remove(app, maps:remove(application_refusal, Request)), Context, Index), Sample#{application => Application}; sample_request(Request, Context, Index) -> sample(Request, Context, Index). distribution(Controller) -> try observer_cli_snapshot:diagnostic_distribution(Controller) of Context -> Context catch _Class:_Reason:_Stacktrace -> #{status => error, reason_code => distribution_probe_failed} end. -ifdef(TEST). sample(#{test_samples := Samples}, _Context, Index) -> lists:nth(Index + 1, Samples); sample(Request, Context, Index) -> observer_cli_snapshot:diagnostic_sample(Request#{sample_index => Index}, Context). -else. sample(Request, Context, Index) -> observer_cli_snapshot:diagnostic_sample(Request#{sample_index => Index}, Context). -endif. sleep_until(Target) -> case Target - erlang:monotonic_time(millisecond) of Remaining when Remaining > 0 -> timer:sleep(Remaining); _ -> ok end. -ifdef(TEST). -spec build_report([map()], [integer()], map(), map()) -> map(). -endif. build_report(Samples, Plan, Timing, Distribution) -> RequiredComplete = required_complete(Samples), ProcessStatus = optional_status(Samples), DistributionStatus = distribution_status(Distribution), EtsContext = current_entity_context(Samples, ets_inventory, [size, memory_words]), PortContext = current_entity_context(Samples, port_inventory, [ queue_size, memory, input, output ]), SchedulerContext = quick_scheduler_context(Samples), ExtraStatuses = [ quick_context_status(Context) || Context <- [ EtsContext, PortContext, SchedulerContext ] ], Status = capture_status(RequiredComplete, [ProcessStatus, DistributionStatus | ExtraStatuses]), RuntimeSamples = runtime_samples(Samples), Findings = case RequiredComplete of true -> limit_findings(RuntimeSamples); false -> [] end, ProcessContext = process_context(Samples), Skipped = skipped_checks(Samples), observer_cli_cli:response( diagnose, Status, #{ node => {identifier, node, node()}, otp_release => unicode:characters_to_binary(erlang:system_info(otp_release)) }, #{ started_at => maps:get(started_at, Timing), finished_at => maps:get(finished_at, Timing), duration_ms => maps:get(duration_ms, Timing), probes => probe_reports(Samples, RequiredComplete, ProcessStatus, DistributionStatus) ++ quick_context_probes(EtsContext, PortContext, SchedulerContext), observer_effects => observer_effects(Timing) }, #{ ruleset => ?RULESET, ruleset_version => ?RULESET_VERSION, sampling_plan => sampling_plan(Plan, Samples), findings => Findings, suspects => [], context => #{ snapshot => #{runtime_samples => RuntimeSamples}, hot_processes_by_reductions => ProcessContext, ets => EtsContext, ports => PortContext, scheduler => SchedulerContext, distribution => Distribution }, skipped => Skipped, summary => summary(Status, Findings) }, [] ). quick_context_status(#{status := Status}) when Status =:= error; Status =:= invalid -> error; quick_context_status(#{status := Status}) when Status =:= ok; Status =:= valid -> ok; quick_context_status(_) -> unavailable. quick_context_probes(Ets, Ports, Scheduler) -> [ quick_context_probe(Id, Context) || {Id, Context} <- [ {ets_inventory, Ets}, {port_inventory, Ports}, {scheduler_pressure, Scheduler} ] ]. quick_context_probe(Id, Context) -> Status = quick_context_status(Context), #{ id => Id, required => false, status => Status, reason_code => maps:get(reason_code, Context, null), duration_ms => 0, samples => case Status of ok -> 1; _ -> 0 end, coverage => [current_context_only] }. current_entity_context(Samples, Field, Metrics) -> case [ Value || Sample <- Samples, Value <- [maps:get(Field, Sample, #{})], maps:get(status, Value, unavailable) =:= ok ] of [#{values := Values} | _] -> Sort = hd(Metrics), Items0 = [ (maps:with(Metrics, Item))#{id => diagnostic_identifier(Id)} || {Id, Item} <- maps:to_list(Values) ], Items = recon_top_n(Items0, Sort, ?CONTEXT_LIMIT), #{status => ok, sort_metric => Sort, sort_semantics => current, items => Items}; [] -> current_field_status(Samples, Field) end. current_field_status(Samples, Field) -> Values = [maps:get(Field, Sample, #{}) || Sample <- Samples], case [Value || #{status := error} = Value <- Values] of [Error | _] -> Error; [] -> case [Value || #{status := unavailable} = Value <- Values] of [Unavailable | _] -> Unavailable; [] -> #{status => unavailable, reason_code => capability_unavailable} end end. quick_scheduler_context(Samples) -> SchedulerSamples = [Scheduler || #{quick_scheduler_sample := Scheduler} <- Samples], case SchedulerSamples of [#{wall_time := undefined}, #{wall_time := undefined}] -> #{status => unavailable, reason_code => scheduler_wall_time_not_enabled}; [First, Second] -> observer_cli_snapshot:diagnostic_scheduler_window(First, Second); _ -> #{status => unavailable, reason_code => scheduler_wall_time_not_enabled} end. observation_report(Mode, Samples, Plan, Holder, Timing, Distribution) -> RequiredComplete = observation_required_complete(Mode, Samples), OptionalStatuses = observation_optional_statuses(Mode, Samples, Holder, Distribution), Status = capture_status(RequiredComplete, OptionalStatuses), RuntimeSamples = runtime_samples(Samples), Windows = scheduler_windows(Samples), Findings = case RequiredComplete of true -> limit_findings(RuntimeSamples) ++ scheduler_findings(Windows); false -> [] end, observer_cli_cli:response( diagnose, Status, #{ node => {identifier, node, node()}, otp_release => unicode:characters_to_binary(erlang:system_info(otp_release)) }, #{ started_at => maps:get(started_at, Timing), finished_at => maps:get(finished_at, Timing), duration_ms => maps:get(duration_ms, Timing), probes => observation_probe_reports(Mode, Samples, Holder, RequiredComplete), observer_effects => [ #{ id => scheduler_wall_time, temporary_enable => true, observer_contaminated => true } | observer_effects(Timing) ] }, #{ ruleset => ruleset(Mode), ruleset_version => ?RULESET_VERSION, sampling_plan => observation_sampling_plan(Mode, Plan, Samples), findings => Findings, suspects => [], context => #{ snapshot => #{runtime_samples => RuntimeSamples}, trends => observation_trends(Samples), scheduler_windows => Windows, application => application_trend(Samples), binary_holders => Holder, distribution => Distribution }, skipped => observation_skipped(Mode, Samples, Holder), summary => observation_summary(Mode, Status, Findings) }, [] ). ruleset(observation) -> <<"observer_cli.observation">>; ruleset(deep) -> <<"observer_cli.deep_observation">>; ruleset(application) -> <<"observer_cli.application_observation">>. observation_required_complete(Mode, Samples) -> Expected = case Mode of deep -> 7; _ -> 5 end, length(Samples) =:= Expected andalso lists:all( fun(Sample) -> valid_required_sample(Sample) andalso valid_memory_sample(Sample) andalso valid_application_sample(Mode, Sample) end, Samples ). valid_memory_sample(#{memory := #{status := ok, values := Values}}) -> is_map(Values) andalso maps:is_key(total_bytes, Values) andalso maps:is_key(binary_bytes, Values); valid_memory_sample(_) -> false. valid_application_sample(application, #{application := #{status := Status}}) -> Status =:= ok orelse Status =:= not_running; valid_application_sample(application, _) -> false; valid_application_sample(_Mode, _Sample) -> true. observation_optional_statuses(Mode, Samples, Holder, Distribution) -> Statuses = [ series_optional_status(Samples, Field) || Field <- [ process_inventory, ets_inventory, port_inventory, socket_inventory ] ] ++ [scheduler_status(scheduler_windows(Samples)), distribution_status(Distribution)], case Mode of deep -> [map_status(Holder) | Statuses]; _ -> Statuses end. series_optional_status(Samples, Field) -> case field_series_status(Samples, Field) of ok -> ok; unavailable -> unavailable; invalid -> error end. map_status(#{status := Status}) when Status =:= error; Status =:= timeout -> error; map_status(#{status := ok}) -> ok; map_status(_) -> unavailable. scheduler_status(Windows) -> case lists:any(fun(#{status := Status}) -> Status =/= valid end, Windows) of true -> error; false -> case Windows of [] -> unavailable; _ -> ok end end. scheduler_windows(Samples) -> scheduler_windows(Samples, []). scheduler_windows([First, Second | Rest], Acc) -> Window = case {maps:find(scheduler_baseline, First), maps:find(scheduler_end, Second)} of {{ok, Baseline}, {ok, End}} -> (observer_cli_snapshot:diagnostic_scheduler_window(Baseline, End))#{ heavy_probe_overlap => false, from_sample_index => length(Acc), to_sample_index => length(Acc) + 1, monotonic_midpoint_ms => maps:get(monotonic_midpoint_ms, Second, 0) }; _ -> #{status => invalid, reason_code => sampling_gap, heavy_probe_overlap => false} end, scheduler_windows([Second | Rest], [Window | Acc]); scheduler_windows(_, Acc) -> lists:reverse(Acc). -ifdef(TEST). -spec scheduler_findings([map()]) -> [map()]. -endif. scheduler_findings(Windows) -> lists:filtermap(fun(Pool) -> scheduler_finding(Pool, Windows) end, [normal, dirty_cpu]). scheduler_finding(Pool, Windows) -> case consecutive_pressure(Pool, Windows, 0, []) of {true, Evidence} -> {true, #{ id => <<"vm.scheduler_pressure">>, severity => warning, entity => #{type => scheduler_pool, id => atom_to_binary(Pool)}, summary => <<"Scheduler pool reached sustained pressure during capture.">>, ruleset_version => ?RULESET_VERSION, evidence => Evidence, recommendations => [ <<"Inspect runnable work and reductions context before tuning schedulers.">> ] }}; false -> false end. consecutive_pressure(_Pool, [], _Run, _Evidence) -> false; consecutive_pressure(Pool, [Window | Rest], Run, Evidence) -> case pressure_window(Pool, Window) of {true, Item} when Run + 1 >= 2 -> {true, lists:reverse([Item | Evidence])}; {true, Item} -> consecutive_pressure(Pool, Rest, Run + 1, [Item | Evidence]); false -> consecutive_pressure(Pool, Rest, 0, []) end. pressure_window( Pool, #{ status := valid, heavy_probe_overlap := false, run_queues := RunQueues } = Window ) -> PoolData = maps:get(Pool, Window, #{}), Queue = maps:get(Pool, RunQueues, #{}), Ratio = maps:get(utilization_ratio, PoolData, 0), Runnable = maps:get(end_observed_runnable_count_including_observer, Queue, 0), case maps:get(status, PoolData, unavailable) =:= available andalso scheduler_over_threshold(PoolData) andalso Runnable > 0 of true -> {true, #{ path => <<"/data/context/scheduler_windows">>, sample_index => maps:get(to_sample_index, Window, 0), monotonic_midpoint_ms => maps:get(monotonic_midpoint_ms, Window, 0), observed => Ratio, operator => <<">=">>, threshold => 0.8, pool => Pool, utilization_ratio => Ratio, observed_runnable_count_including_observer => Runnable, wall_time_unit => opaque_same_window, run_queue_snapshot_atomic => false, heavy_probe_overlap => false }}; false -> false end; pressure_window(_Pool, _Window) -> false. scheduler_over_threshold(#{active_delta := #{value := Active}, total_delta := #{value := Total}}) when is_integer(Active), is_integer(Total), Total > 0 -> Active * 100 >= Total * 80; scheduler_over_threshold(#{utilization_ratio := Ratio}) -> Ratio >= 0.8; scheduler_over_threshold(_) -> false. observation_trends(Samples) -> case lists:all(fun(Sample) -> maps:get(status, Sample, error) =:= ok end, Samples) of false -> #{status => invalid, reason_code => sampling_gap}; true -> valid_observation_trends(Samples) end. valid_observation_trends(Samples) -> #{ status => ok, global_memory => map_gauge_trend(Samples, memory, values), processes => entity_trends( Samples, process_inventory, values, [message_queue_len, memory_bytes] ), ets => entity_trends(Samples, ets_inventory, values, [size, memory_words]), ports => entity_trends(Samples, port_inventory, values, [queue_size, memory, input, output]), sockets => socket_trends(Samples) }. map_gauge_trend([], _Field, _Values) -> #{status => unavailable}; map_gauge_trend(Samples, Field, ValuesKey) -> First = maps:get(ValuesKey, maps:get(Field, hd(Samples), #{}), #{}), Last = maps:get(ValuesKey, maps:get(Field, lists:last(Samples), #{}), #{}), Interval = inventory_interval(Samples, Field), Deltas = gauge_deltas(First, Last), #{ status => ok, sample_count => length(Samples), interval_ms => Interval, deltas => Deltas, rates_per_second => rates(Deltas, Interval) }. gauge_deltas(First, Last) -> maps:from_list([ {Key, maps:get(Key, Last) - maps:get(Key, First)} || Key <- maps:keys(First), is_integer(maps:get(Key, First)), is_integer(maps:get(Key, Last, undefined)) ]). entity_trends([], _Field, _Values, _Metrics) -> #{status => unavailable, items => []}; entity_trends(Samples, Field, ValuesKey, Metrics) -> case field_series_status(Samples, Field) of ok -> valid_entity_trends(Samples, Field, ValuesKey, Metrics); unavailable -> #{status => unavailable, items => []}; invalid -> #{status => invalid, reason_code => sampling_gap, items => []} end. valid_entity_trends(Samples, Field, ValuesKey, Metrics) -> ValueMaps = [maps:get(ValuesKey, maps:get(Field, Sample, #{}), #{}) || Sample <- Samples], First = hd(ValueMaps), Last = lists:last(ValueMaps), Shared = lists:foldl( fun(Values, Ids) -> ordsets:intersection(Ids, lists:sort(maps:keys(Values))) end, lists:sort(maps:keys(First)), tl(ValueMaps) ), Stable = [Id || Id <- Shared, same_generation([maps:get(Id, Values) || Values <- ValueMaps])], Replaced = ordsets:subtract(Shared, Stable), Interval = inventory_interval(Samples, Field), Items0 = [ entity_trend_item(Id, [maps:get(Id, Values) || Values <- ValueMaps], Metrics, Interval) || Id <- Stable ], SortMetric = hd(Metrics), Items = recon_top_n(Items0, SortMetric, ?CONTEXT_LIMIT), #{ status => ok, sample_count => length(Samples), born_count => length( ordsets:subtract(lists:sort(maps:keys(Last)), lists:sort(maps:keys(First))) ), dead_count => length( ordsets:subtract(lists:sort(maps:keys(First)), lists:sort(maps:keys(Last))) ), replaced_count => length(Replaced), replaced => [diagnostic_identifier(Id) || Id <- Replaced], interval_ms => Interval, sort_metric => SortMetric, sort_semantics => delta_descending, items => [maps:remove(raw_id, Item) || Item <- Items] }. field_series_status(Samples, Field) -> Statuses = [maps:get(status, maps:get(Field, Sample, #{}), missing) || Sample <- Samples], case { lists:all(fun(Status) -> Status =:= ok end, Statuses), lists:all(fun(Status) -> Status =:= unavailable end, Statuses) } of {true, _} -> ok; {_, true} -> unavailable; _ -> invalid end. socket_trends(Samples) -> case field_series_status(Samples, socket_inventory) of ok -> observer_cli_snapshot:diagnostic_socket_trend([ maps:get(values, maps:get(socket_inventory, Sample)) || Sample <- Samples ]); unavailable -> #{status => unavailable, items => []}; invalid -> #{status => invalid, reason_code => sampling_gap, items => []} end. same_generation([#{generation := Generation} | Rest]) -> lists:all( fun (#{generation := ItemGeneration}) -> ItemGeneration =:= Generation; (_) -> false end, Rest ); same_generation(_Items) -> true. entity_trend_item(Id, Values, Metrics, Interval) -> Deltas = metric_series_deltas(Values, Metrics), Base = #{ raw_id => Id, id => diagnostic_identifier(Id), deltas => Deltas, rates_per_second => rates(Deltas, Interval) }, case lists:member(message_queue_len, Metrics) of true -> Base#{ message_queue_positive_step_ratio => positive_step_ratio( [maps:get(message_queue_len, Value, undefined) || Value <- Values] ) }; false -> Base end. sample_interval(Samples) -> maps:get(monotonic_midpoint_ms, lists:last(Samples), 0) - maps:get(monotonic_midpoint_ms, hd(Samples), 0). inventory_interval(Samples, Field) -> case { inventory_midpoint(hd(Samples), Field), inventory_midpoint(lists:last(Samples), Field) } of {First, Last} when is_integer(First), is_integer(Last) -> Last - First; _ -> sample_interval(Samples) end. inventory_midpoint(Sample, Field) -> Audit = maps:get(audit, maps:get(Field, Sample, #{}), #{}), case { maps:get(scan_started_monotonic_ms, Audit, undefined), maps:get(scan_finished_monotonic_ms, Audit, undefined) } of {Started, Finished} when is_integer(Started), is_integer(Finished) -> Started + ((Finished - Started) div 2); _ -> undefined end. reductions_rates(Items, Interval) when Interval > 0 -> [ case maps:get(reductions_delta, Item, null) of Delta when is_integer(Delta) -> Item#{reductions_per_second => Delta * 1000 / Interval}; _ -> Item#{reductions_per_second => null} end || Item <- Items ]; reductions_rates(Items, _Interval) -> [Item#{reductions_per_second => null} || Item <- Items]. rates(Deltas, Interval) when Interval > 0 -> maps:from_list([ {Key, Value * 1000 / Interval} || {Key, Value} <- maps:to_list(Deltas), is_integer(Value) ]); rates(_Deltas, _Interval) -> #{}. positive_step_ratio(Values) -> Steps = [ {A, B} || {A, B} <- lists:zip(lists:droplast(Values), tl(Values)), is_integer(A), is_integer(B) ], case Steps of [] -> null; _ -> length([ok || {A, B} <- Steps, B > A]) / length(Steps) end. diagnostic_identifier(Id) when is_pid(Id) -> {identifier, pid, Id}; diagnostic_identifier(Id) when is_port(Id) -> {identifier, port, Id}; diagnostic_identifier(Id) -> {identifier, table, Id}. metric_series_deltas(Values, Metrics) -> lists:foldl(fun(Key, Acc) -> metric_series_delta(Key, Values, Acc) end, #{}, Metrics). metric_series_delta(Key, Values, Acc) when Key =:= input; Key =:= output -> Series = [maps:get(Key, Value, undefined) || Value <- Values], case lists:all(fun is_integer/1, Series) of true when length(Series) >= 2 -> case lists:any( fun({Before, After}) -> After < Before end, lists:zip(lists:droplast(Series), tl(Series)) ) of true -> Acc#{ Key => null, iolist_to_binary([atom_to_binary(Key), <<"_state">>]) => counter_reset }; false -> Acc#{Key => lists:last(Series) - hd(Series)} end; _ -> Acc end; metric_series_delta(Key, Values, Acc) -> case {maps:get(Key, hd(Values), undefined), maps:get(Key, lists:last(Values), undefined)} of {Before, After} when is_integer(Before), is_integer(After) -> Acc#{Key => After - Before}; _ -> Acc end. -ifdef(TEST). -spec application_trend([map()]) -> map(). -endif. application_trend(Samples) -> AppSamples = [maps:get(application, Sample, #{}) || Sample <- Samples], Statuses = [maps:get(status, App, unavailable) || App <- AppSamples], ScanRefused = lists:any( fun(App) -> maps:get(reason_code, App, undefined) =:= scan_budget_exceeded end, AppSamples ), case { ScanRefused, lists:all(fun(Status) -> Status =:= ok orelse Status =:= not_running end, Statuses), lists:all(fun(Status) -> Status =:= unavailable end, Statuses) } of {true, _Complete, _Unavailable} -> #{status => unavailable, reason_code => scan_budget_exceeded, items => []}; {false, false, true} -> #{status => unavailable, items => []}; {false, false, false} -> #{status => invalid, reason_code => sampling_gap, items => []}; {false, true, _} -> ChildSamples = [application_children(App) || App <- AppSamples], First = application_children(hd(AppSamples)), Last = application_children(lists:last(AppSamples)), Stable = lists:foldl( fun(Children, Ids) -> ordsets:intersection(Ids, lists:sort(maps:keys(Children))) end, lists:sort(maps:keys(First)), tl(ChildSamples) ), Items = [ #{ id => Id, pid_changed => case lists:usort([maps:get(Id, Children) || Children <- ChildSamples]) of [_First, _Second | _Rest] -> true; _ -> false end, change_semantics => restart_deploy_or_manual_change } || Id <- Stable ], #{ status => ok, sample_count => length(AppSamples), items => Items, born_count => maps:size(Last) - length(Stable), dead_count => maps:size(First) - length(Stable), identity_unavailable_count => lists:sum([ maps:get(identity_unavailable_count, A, 0) || A <- AppSamples ]) } end. application_children(App) -> maps:from_list([ {Id, maps:get(pid, Child, null)} || #{identity := available, id := Id, child := Child} <- maps:get(children, App, []) ]). observation_sampling_plan(Mode, Plan, Samples) -> #{ mode => Mode, planned_sample_count => length(Plan), planned_duration_ms => lists:last(Plan) - hd(Plan), target_monotonic_times_ms => Plan, actual_samples => [ #{ sample_index => Index, target_monotonic_ms => lists:nth(Index + 1, Plan), started_monotonic_ms => maps:get(monotonic_start_ms, Sample, null), finished_monotonic_ms => maps:get(monotonic_finish_ms, Sample, null), status => maps:get(status, Sample, error), reason_code => maps:get(reason_code, Sample, null) } || {Index, Sample} <- indexed(Samples) ] }. observation_probe_reports(Mode, Samples, Holder, RequiredComplete) -> [ #{ id => core_limits_and_memory, required => true, status => status(RequiredComplete), reason_code => reason(RequiredComplete, required_coverage_incomplete), duration_ms => sample_duration(Samples), samples => length([ ok || S <- Samples, valid_required_sample(S) andalso valid_memory_sample(S) ]), coverage => [ target_identity, process_count_limit, port_count_limit, atom_count_limit, ets_count_limit, global_memory ] }, #{ id => scheduler_pressure, required => false, status => scheduler_status(scheduler_windows(Samples)), reason_code => scheduler_reason(scheduler_windows(Samples)), duration_ms => sample_duration(Samples), samples => length(scheduler_windows(Samples)), coverage => [low_cost_windows, online_topology, opaque_same_window_units] }, #{ id => binary_holders, required => false, status => case Mode of deep -> map_status(Holder); _ -> unavailable end, reason_code => maps:get(reason_code, Holder, null), duration_ms => 0, samples => case maps:get(status, Holder, unavailable) of ok -> 1; _ -> 0 end, coverage => [final_independent_admission, current_context_only] } ] ++ [ observation_inventory_probe(Field, Samples) || Field <- [ process_inventory, ets_inventory, port_inventory, socket_inventory ] ]. observation_inventory_probe(Field, Samples) -> case field_series_status(Samples, Field) of ok -> #{ id => Field, required => false, status => ok, reason_code => null, duration_ms => sample_duration(Samples), samples => length(Samples), coverage => [exact_generation_trend] }; unavailable -> #{ id => Field, required => false, status => unavailable, reason_code => first_field_reason(Samples, Field), duration_ms => 0, samples => 0, coverage => [] }; invalid -> #{ id => Field, required => false, status => error, reason_code => sampling_gap, duration_ms => sample_duration(Samples), samples => length(Samples), coverage => [] } end. first_field_reason(Samples, Field) -> case [ Reason || Sample <- Samples, #{reason_code := Reason} <- [maps:get(Field, Sample, #{})] ] of [Reason | _] -> Reason; [] -> capability_unavailable end. scheduler_reason(Windows) -> case [ maps:get(reason_code, Window, scheduler_window_invalid) || Window <- Windows, maps:get(status, Window, invalid) =/= valid ] of [Reason | _] -> Reason; [] -> null end. observation_skipped(Mode, _Samples, _Holder) -> Growth = [ mailbox_backlog_suspects, memory_growth_suspects, ets_growth_suspects, port_queue_backlog_suspects, supervisor_restart_suspects ], Binary = case Mode of deep -> [#{id => binary_retention_suspects, reason_code => ruleset_not_calibrated}]; _ -> [#{id => binary_retention_suspects, reason_code => deep_not_requested}] end, [#{id => Id, reason_code => ruleset_not_calibrated} || Id <- Growth] ++ Binary. observation_summary(Mode, partial, _Findings) -> iolist_to_binary( io_lib:format("~p diagnostics capture is partial; findings suppressed.", [Mode]) ); observation_summary(Mode, complete, Findings) -> iolist_to_binary( io_lib:format("~p diagnostics completed with ~B finding(s).", [Mode, length(Findings)]) ). required_complete([_, _] = Samples) -> lists:all(fun valid_required_sample/1, Samples); required_complete(_Samples) -> false. valid_required_sample(#{status := ok, resources := Resources}) -> lists:all( fun(Domain) -> valid_limit(maps:get(Domain, Resources, invalid)) end, [process, port, atom, ets] ); valid_required_sample(_Sample) -> false. valid_limit(#{observed_count_including_observer := Count, limit := Limit}) -> is_integer(Count) andalso Count >= 0 andalso is_integer(Limit) andalso Limit > 0; valid_limit(#{observed_count := Count, limit := Limit}) -> is_integer(Count) andalso Count >= 0 andalso is_integer(Limit) andalso Limit > 0; valid_limit(_Value) -> false. runtime_samples(Samples) -> [ runtime_sample(Index, Sample) || {Index, Sample} <- indexed(Samples), valid_required_sample(Sample) ]. runtime_sample(Index, #{resources := Resources} = Sample) -> lists:foldl( fun(Domain, Acc) -> Limit = maps:get(Domain, Resources), Count = observed_count(Limit), Acc#{ count_key(Domain) => Count, limit_key(Domain) => maps:get(limit, Limit), ratio_key(Domain) => Count / maps:get(limit, Limit) } end, #{ sample_index => Index, monotonic_start_ms => maps:get(monotonic_start_ms, Sample), monotonic_finish_ms => maps:get(monotonic_finish_ms, Sample), monotonic_midpoint_ms => maps:get(monotonic_midpoint_ms, Sample) }, [process, port, atom, ets] ). observed_count(#{observed_count_including_observer := Count}) -> Count; observed_count(#{observed_count := Count}) -> Count. -ifdef(TEST). -spec limit_findings([map()]) -> [map()]. -endif. limit_findings(RuntimeSamples) -> lists:filtermap(fun(Domain) -> limit_finding(Domain, RuntimeSamples) end, [ process, port, atom, ets ]). limit_finding(Domain, Samples) -> Highest = lists:foldl( fun(Sample, Best) -> higher_sample(Domain, Sample, Best) end, none, Samples ), case Highest of none -> false; Sample -> Count = maps:get(count_key(Domain), Sample), Limit = maps:get(limit_key(Domain), Sample), case severity(Count, Limit) of none -> false; Severity -> {true, finding(Domain, Severity, Count, Limit, Sample)} end end. higher_sample(_Domain, Sample, none) -> Sample; higher_sample(Domain, Sample, Best) -> Count = maps:get(count_key(Domain), Sample), Limit = maps:get(limit_key(Domain), Sample), BestCount = maps:get(count_key(Domain), Best), BestLimit = maps:get(limit_key(Domain), Best), case Count * BestLimit > BestCount * Limit of true -> Sample; false -> Best end. severity(Count, Limit) when Count * 100 >= Limit * 95 -> critical; severity(Count, Limit) when Count * 100 > Limit * 85 -> warning; severity(_Count, _Limit) -> none. finding(Domain, Severity, Count, Limit, Sample) -> Index = maps:get(sample_index, Sample), Ratio = Count / Limit, #{ id => rule_id(Domain), severity => Severity, entity => #{type => node, id => {identifier, node, node()}}, summary => iolist_to_binary( io_lib:format( "~s table reached ~.1f% during capture.", [domain_name(Domain), Ratio * 100] ) ), ruleset_version => ?RULESET_VERSION, evidence => [ #{ path => iolist_to_binary( io_lib:format( "/data/context/snapshot/runtime_samples/~B/~s_usage_ratio", [Index, atom_to_list(Domain)] ) ), sample_index => Index, monotonic_midpoint_ms => maps:get(monotonic_midpoint_ms, Sample), observed => Ratio, observed_count_including_observer => Count, limit => Limit, observer_contaminated => Domain =/= ets, operator => operator(Severity), threshold => threshold(Severity) } ], recommendations => [recommendation(Domain)] }. operator(critical) -> <<">=">>; operator(warning) -> <<">">>. threshold(critical) -> 0.95; threshold(warning) -> 0.85. rule_id(process) -> <<"vm.process_limit_pressure">>; rule_id(port) -> <<"vm.port_limit_pressure">>; rule_id(atom) -> <<"vm.atom_limit_pressure">>; rule_id(ets) -> <<"vm.ets_limit_pressure">>. domain_name(process) -> "Process"; domain_name(port) -> "Port"; domain_name(atom) -> "Atom"; domain_name(ets) -> "ETS". recommendation(process) -> <<"Inspect process memory and message-queue rankings before changing the limit.">>; recommendation(port) -> <<"Inspect Port ownership and queues before changing the limit.">>; recommendation(atom) -> <<"Inspect atom creation paths before changing the limit.">>; recommendation(ets) -> <<"Inspect ETS ownership and table counts before changing the limit.">>. process_context([First, Second]) -> case {inventory(First), inventory(Second)} of {{ok, FirstInventory}, {ok, SecondInventory}} -> Interval = inventory_interval([First, Second], process_inventory), Context = reductions_context(FirstInventory, SecondInventory), Context#{ interval_ms => Interval, items => reductions_rates(maps:get(items, Context), Interval) }; {{unavailable, Reason}, _} -> unavailable_context(Reason); {_, {unavailable, Reason}} -> unavailable_context(Reason); _ -> #{status => partial, reason_code => process_inventory_failed, items => []} end; process_context(_Samples) -> #{status => partial, reason_code => sampling_gap, items => []}. inventory(#{process_inventory := #{status := ok, values := Values} = Inventory}) -> {ok, Inventory#{values := Values}}; inventory(#{process_inventory := #{status := unavailable, reason_code := Reason}}) -> {unavailable, Reason}; inventory(#{process_inventory := #{status := Status}}) when Status =:= error; Status =:= timeout -> {error, Status}; inventory(_Sample) -> {unavailable, capability_unavailable}. unavailable_context(Reason) -> #{status => unavailable, reason_code => Reason, items => []}. -ifdef(TEST). -spec reductions_context(map(), map()) -> map(). -endif. reductions_context(#{values := First} = FirstInventory, #{values := Second} = SecondInventory) -> FirstIds = lists:sort(maps:keys(First)), SecondIds = lists:sort(maps:keys(Second)), Stable = ordsets:intersection(FirstIds, SecondIds), Born = ordsets:subtract(SecondIds, FirstIds), Dead = ordsets:subtract(FirstIds, SecondIds), Items0 = [process_delta(Pid, maps:get(Pid, First), maps:get(Pid, Second)) || Pid <- Stable], Reset = [ Pid || #{pid := {identifier, pid, Pid}, reductions_state := counter_reset} <- Items0 ], {BornPids, BornPidsTruncated} = pid_sample(Born, ?CONTEXT_LIMIT), {DeadPids, DeadPidsTruncated} = pid_sample(Dead, ?CONTEXT_LIMIT), {ResetPids, ResetPidsTruncated} = pid_sample(Reset, ?CONTEXT_LIMIT), Denominator = lists:sum([ Delta || #{reductions_delta := Delta} <- Items0, is_integer(Delta), Delta > 0 ]), Items1 = [add_share(Item, Denominator) || Item <- Items0], Items = recon_top_n(Items1, reductions_delta, ?CONTEXT_LIMIT), #{ status => ok, denominator_semantics => all_stable_scanned_positive_reductions, stable_positive_reductions_denominator => Denominator, born_count => length(Born), dead_count => length(Dead), reset_count => length(Reset), born_pids => BornPids, born_pids_truncated => BornPidsTruncated, dead_pids => DeadPids, dead_pids_truncated => DeadPidsTruncated, reset_pids => ResetPids, reset_pids_truncated => ResetPidsTruncated, sample_audits => [ maps:get(audit, Inventory, #{}) || Inventory <- [ FirstInventory, SecondInventory ] ], items => Items }. pid_sample(Pids, Limit) -> {[{identifier, pid, Pid} || Pid <- lists:sublist(Pids, Limit)], length(Pids) > Limit}. process_delta(Pid, First, Second) -> Before = maps:get(reductions, First), After = maps:get(reductions, Second), Reduction = case After >= Before of true -> #{reductions_state => available, reductions_delta => After - Before}; false -> #{reductions_state => counter_reset, reductions_delta => null} end, Reduction#{ pid => {identifier, pid, Pid}, message_queue_len_delta => maps:get(message_queue_len, Second) - maps:get(message_queue_len, First), memory_bytes_delta => maps:get(memory_bytes, Second) - maps:get(memory_bytes, First) }. add_share(#{reductions_delta := Delta} = Item, Denominator) when is_integer(Delta), Delta > 0, Denominator > 0 -> Item#{share_of_stable_scanned_reductions => Delta / Denominator}; add_share(Item, _Denominator) -> Item#{share_of_stable_scanned_reductions => null}. optional_status(Samples) -> Inventories = [inventory(Sample) || Sample <- Samples], case lists:any( fun ({error, _}) -> true; (_) -> false end, Inventories ) of true -> error; false -> case lists:all( fun ({ok, _}) -> true; (_) -> false end, Inventories ) of true -> ok; false -> unavailable end end. distribution_status(#{status := error}) -> error; distribution_status(_Distribution) -> ok. capture_status(true, OptionalStatuses) -> case lists:member(error, OptionalStatuses) of true -> partial; false -> complete end; capture_status(false, _OptionalStatuses) -> partial. probe_reports(Samples, RequiredComplete, ProcessStatus, DistributionStatus) -> [ #{ id => core_limits, required => true, status => status(RequiredComplete), reason_code => reason(RequiredComplete, required_coverage_incomplete), duration_ms => sample_duration(Samples), samples => length([ok || Sample <- Samples, valid_required_sample(Sample)]), coverage => [process_count_limit, port_count_limit, atom_count_limit, ets_count_limit] }, #{ id => process_inventory, required => false, status => ProcessStatus, reason_code => optional_reason(ProcessStatus, Samples), duration_ms => inventory_duration(Samples), samples => length([ok || Sample <- Samples, element(1, inventory(Sample)) =:= ok]), coverage => [shared_same_point_process_facts, stable_pid_intersection] }, #{ id => distribution, required => false, status => DistributionStatus, reason_code => reason(DistributionStatus =:= ok, distribution_probe_failed), duration_ms => 0, samples => 1, coverage => [public_connected_peers, context_only] } ]. status(true) -> ok; status(false) -> error. reason(true, _Reason) -> null; reason(false, Reason) -> Reason. optional_reason(ok, _Samples) -> null; optional_reason(error, _Samples) -> process_inventory_failed; optional_reason(unavailable, Samples) -> case [Reason || Sample <- Samples, {unavailable, Reason} <- [inventory(Sample)]] of [Reason | _] -> Reason; [] -> capability_unavailable end. sample_duration([]) -> 0; sample_duration(Samples) -> Starts = [maps:get(monotonic_start_ms, S) || S <- Samples, maps:is_key(monotonic_start_ms, S)], Finishes = [ maps:get(monotonic_finish_ms, S) || S <- Samples, maps:is_key(monotonic_finish_ms, S) ], case {Starts, Finishes} of {[], _} -> 0; {_, []} -> 0; _ -> lists:max(Finishes) - lists:min(Starts) end. inventory_duration(Samples) -> lists:sum([ maps:get(scan_finished_monotonic_ms, Audit, 0) - maps:get(scan_started_monotonic_ms, Audit, 0) || Sample <- Samples, #{process_inventory := #{audit := Audit}} <- [Sample] ]). sampling_plan(Plan, Samples) -> #{ mode => quick, planned_sample_count => 2, planned_interval_ms => lists:nth(2, Plan) - lists:nth(1, Plan), target_monotonic_times_ms => Plan, actual_samples => [ #{ sample_index => Index, target_monotonic_ms => lists:nth(Index + 1, Plan), started_monotonic_ms => maps:get(monotonic_start_ms, Sample, null), finished_monotonic_ms => maps:get(monotonic_finish_ms, Sample, null), midpoint_monotonic_ms => maps:get(monotonic_midpoint_ms, Sample, null) } || {Index, Sample} <- indexed(Samples) ] }. skipped_checks(_Samples) -> Growth = [ mailbox_backlog_suspects, memory_growth_suspects, ets_growth_suspects, port_queue_backlog_suspects, binary_retention_suspects ], [#{id => Id, reason_code => ruleset_not_calibrated} || Id <- Growth]. summary(partial, _Findings) -> <<"Quick diagnostics capture is partial; findings suppressed.">>; summary(complete, []) -> <<"Quick diagnostics completed with no limit findings.">>; summary(complete, Findings) -> iolist_to_binary( io_lib:format("Quick diagnostics found ~B limit finding(s).", [length(Findings)]) ). observer_effects(Timing) -> [ #{ id => diagnostics_worker, affected_facts => [process_count, port_count, atom_count, memory, reductions] }, #{ id => module_load, module_loaded_before_sample => maps:get(module_loaded_before_sample, Timing) }, #{ id => distribution_controller, controller_peer => {identifier, peer, node(maps:get(controller, Timing))}, dynamic_controller_name_atom => true } ]. count_key(process) -> process_observed_count_including_observer; count_key(port) -> port_observed_count_including_observer; count_key(atom) -> atom_observed_count_including_observer; count_key(ets) -> ets_observed_count_including_observer. limit_key(process) -> process_limit; limit_key(port) -> port_limit; limit_key(atom) -> atom_limit; limit_key(ets) -> ets_limit. ratio_key(process) -> process_usage_ratio; ratio_key(port) -> port_usage_ratio; ratio_key(atom) -> atom_usage_ratio; ratio_key(ets) -> ets_usage_ratio. recon_top_n(Items, Sort, Limit) -> [ Item || {_, _, [Item]} <- recon_lib:sublist_top_n_attrs( [{top_n_identity(Item), top_n_value(Item, Sort), [Item]} || Item <- Items], Limit ) ]. top_n_identity(#{pid := {identifier, pid, Pid}}) -> Pid; top_n_identity(_Item) -> 0. top_n_value(#{deltas := Deltas}, Sort) -> maps:get(Sort, Deltas, -1); top_n_value(Item, Sort) -> case maps:get(Sort, Item, null) of Value when is_number(Value) -> Value; _ -> -1 end. indexed(List) -> indexed(List, 0). indexed([Item | Rest], Index) -> [{Index, Item} | indexed(Rest, Index + 1)]; indexed([], _Index) -> []. rfc3339(SystemTime) -> unicode:characters_to_binary( calendar:system_time_to_rfc3339(SystemTime, [{unit, millisecond}, {offset, "Z"}]) ).