%% @doc System resource monitoring for adaptive neuroevolution. %% %% This module provides system resource metrics that can be used as inputs %% to the LTC meta-controller, allowing it to adapt evolution hyperparameters %% based on current memory and CPU pressure. %% %% == Metrics Provided == %% %% Raw metrics: %% - memory_total: Total VM memory usage (bytes) %% - memory_processes: Memory used by processes (bytes) %% - memory_binary: Binary memory (bytes) - often the culprit in leaks %% - scheduler_utilization: Average CPU utilization (0.0 - 1.0) %% - process_count: Number of processes in the VM %% - message_queue_len: Total message queue length across monitored processes %% %% Normalized metrics (for LTC input): %% - memory_pressure: 0.0 (plenty of memory) to 1.0 (critical) %% - cpu_pressure: 0.0 (idle) to 1.0 (saturated) %% - process_pressure: 0.0 (few processes) to 1.0 (at limit) %% %% == Usage == %% %% %% Get raw metrics %% #{memory_total := Mem} = resource_monitor:get_metrics(), %% %% %% Get normalized metrics for LTC input %% #{memory_pressure := MemP, cpu_pressure := CpuP} = resource_monitor:get_normalized_metrics(), %% LtcInputs = [MemP, CpuP, ...] %% %% @author Macula.io %% @copyright 2025 Macula.io -module(resource_monitor). -export([ get_metrics/0, get_normalized_metrics/0, get_memory_limit/0, is_memory_critical/0, is_memory_high/0, check_health/0 ]). %%% ============================================================================ %%% API Functions %%% ============================================================================ %% @doc Get current system resource metrics (raw values). %% %% Returns a map with various resource measurements. -spec get_metrics() -> map(). get_metrics() -> Memory = erlang:memory(), #{ memory_total => proplists:get_value(total, Memory, 0), memory_processes => proplists:get_value(processes, Memory, 0), memory_processes_used => proplists:get_value(processes_used, Memory, 0), memory_binary => proplists:get_value(binary, Memory, 0), memory_ets => proplists:get_value(ets, Memory, 0), memory_atom => proplists:get_value(atom, Memory, 0), scheduler_utilization => get_scheduler_utilization(), process_count => erlang:system_info(process_count), process_limit => erlang:system_info(process_limit), message_queue_len => get_sampled_message_queue_len(), run_queue => erlang:statistics(run_queue) }. %% @doc Get normalized metrics for LTC input (0.0 to 1.0 range). %% %% These are suitable for direct use as neural network inputs. -spec get_normalized_metrics() -> map(). get_normalized_metrics() -> Metrics = get_metrics(), MaxMemory = get_memory_limit(), ProcessLimit = maps:get(process_limit, Metrics, 262144), MemoryTotal = maps:get(memory_total, Metrics, 0), SchedulerUtil = maps:get(scheduler_utilization, Metrics, 0.0), ProcessCount = maps:get(process_count, Metrics, 0), RunQueue = maps:get(run_queue, Metrics, 0), MessageQueueLen = maps:get(message_queue_len, Metrics, 0), %% Memory pressure: ratio of used to available memory MemoryPressure = min(1.0, MemoryTotal / MaxMemory), %% CPU pressure: combination of scheduler utilization and run queue %% Run queue > 0 indicates backed-up work RunQueuePressure = min(1.0, RunQueue / max(1, erlang:system_info(schedulers))), CpuPressure = (SchedulerUtil + RunQueuePressure) / 2.0, %% Process pressure: ratio of active to max processes ProcessPressure = min(1.0, ProcessCount / ProcessLimit), %% Message queue pressure: high queue lengths indicate backpressure %% Normalize assuming 10000 total queued messages is "full" MessageQueuePressure = min(1.0, MessageQueueLen / 10000.0), #{ memory_pressure => MemoryPressure, cpu_pressure => CpuPressure, process_pressure => ProcessPressure, message_queue_pressure => MessageQueuePressure, %% Composite pressure metric (for simple decisions) overall_pressure => (MemoryPressure + CpuPressure + ProcessPressure) / 3.0 }. %% @doc Get the memory limit for the system. %% %% Attempts to detect from: %% 1. MACULA_MEMORY_LIMIT environment variable (bytes) %% 2. Container cgroup limits %% 3. System total memory %% 4. Default fallback (8GB) -spec get_memory_limit() -> pos_integer(). get_memory_limit() -> case os:getenv("MACULA_MEMORY_LIMIT") of false -> detect_memory_limit(); EnvValue -> try list_to_integer(EnvValue) catch _:_ -> detect_memory_limit() end end. %% @doc Check if memory usage is at critical level (>90%). -spec is_memory_critical() -> boolean(). is_memory_critical() -> #{memory_pressure := Pressure} = get_normalized_metrics(), Pressure > 0.9. %% @doc Check if memory usage is at high level (>70%). -spec is_memory_high() -> boolean(). is_memory_high() -> #{memory_pressure := Pressure} = get_normalized_metrics(), Pressure > 0.7. %% @doc Perform health check and return status. %% %% Returns a map with health status and any warnings/alerts. -spec check_health() -> map(). check_health() -> Metrics = get_normalized_metrics(), MemoryPressure = maps:get(memory_pressure, Metrics), CpuPressure = maps:get(cpu_pressure, Metrics), MsgPressure = maps:get(message_queue_pressure, Metrics), Status = if MemoryPressure > 0.9 -> critical; MemoryPressure > 0.7 orelse CpuPressure > 0.9 -> warning; MsgPressure > 0.8 -> degraded; true -> healthy end, Warnings = lists:filtermap( fun({_Metric, Value, Threshold, Msg}) -> case Value > Threshold of true -> {true, Msg}; false -> false end end, [ {memory, MemoryPressure, 0.7, <<"Memory usage high">>}, {memory, MemoryPressure, 0.9, <<"Memory usage critical">>}, {cpu, CpuPressure, 0.9, <<"CPU saturated">>}, {queue, MsgPressure, 0.8, <<"Message queues backed up">>} ] ), #{ status => Status, warnings => Warnings, metrics => Metrics }. %%% ============================================================================ %%% Internal Functions %%% ============================================================================ %% @private Detect memory limit from system or cgroups. detect_memory_limit() -> %% Try cgroup v2 first (container environments) case file:read_file("/sys/fs/cgroup/memory.max") of {ok, Bin} -> case parse_cgroup_limit(Bin) of max -> detect_system_memory(); Limit -> Limit end; _ -> %% Try cgroup v1 case file:read_file("/sys/fs/cgroup/memory/memory.limit_in_bytes") of {ok, Bin} -> case parse_cgroup_limit(Bin) of max -> detect_system_memory(); Limit -> Limit end; _ -> detect_system_memory() end end. %% @private Parse cgroup limit value. parse_cgroup_limit(Bin) -> Str = string:trim(binary_to_list(Bin)), case Str of "max" -> max; _ -> try list_to_integer(Str) catch _:_ -> max end end. %% @private Detect system memory from /proc/meminfo. detect_system_memory() -> case file:read_file("/proc/meminfo") of {ok, Bin} -> Lines = string:split(binary_to_list(Bin), "\n", all), parse_meminfo(Lines); _ -> %% Default fallback: 8GB 8 * 1024 * 1024 * 1024 end. %% @private Parse MemTotal from /proc/meminfo. parse_meminfo([]) -> 8 * 1024 * 1024 * 1024; % Default fallback parse_meminfo(["MemTotal:" ++ Rest | _]) -> case string:tokens(string:trim(Rest), " \t") of [NumStr, "kB" | _] -> try list_to_integer(NumStr) * 1024 catch _:_ -> 8 * 1024 * 1024 * 1024 end; _ -> 8 * 1024 * 1024 * 1024 end; parse_meminfo([_ | Rest]) -> parse_meminfo(Rest). %% @private Get scheduler utilization. %% %% Uses erlang:statistics(scheduler_wall_time) if available, %% otherwise returns a simple approximation based on run queue. get_scheduler_utilization() -> try %% scheduler_wall_time needs to be enabled first case erlang:statistics(scheduler_wall_time) of undefined -> %% Not enabled, try to enable and return approximation erlang:system_flag(scheduler_wall_time, true), estimate_cpu_from_run_queue(); Times when is_list(Times) -> calculate_scheduler_utilization(Times) end catch _:_ -> estimate_cpu_from_run_queue() end. %% @private Calculate utilization from wall time samples. calculate_scheduler_utilization(Times) -> %% Times = [{SchedulerId, ActiveTime, TotalTime}] {TotalActive, TotalWall} = lists:foldl( fun({_Id, Active, Wall}, {AccActive, AccWall}) -> {AccActive + Active, AccWall + Wall} end, {0, 0}, Times ), case TotalWall of 0 -> 0.0; _ -> min(1.0, TotalActive / TotalWall) end. %% @private Estimate CPU usage from run queue. estimate_cpu_from_run_queue() -> RunQueue = erlang:statistics(run_queue), Schedulers = erlang:system_info(schedulers), %% If run queue is at scheduler count, we're at 100% min(1.0, RunQueue / Schedulers). %% @private Get sampled message queue length. %% %% Samples a subset of processes to estimate total queue length. %% Full enumeration would be too expensive. get_sampled_message_queue_len() -> %% Sample up to 100 random processes AllProcs = erlang:processes(), SampleSize = min(100, length(AllProcs)), Sampled = sample_random(AllProcs, SampleSize), %% Sum message queue lengths Total = lists:foldl( fun(Pid, Acc) -> case erlang:process_info(Pid, message_queue_len) of {message_queue_len, Len} -> Acc + Len; undefined -> Acc end end, 0, Sampled ), %% Extrapolate to full process count if we sampled case {SampleSize, length(AllProcs)} of {S, T} when S < T -> round(Total * T / S); _ -> Total end. %% @private Sample N random elements from list. sample_random(List, N) when N >= length(List) -> List; sample_random(List, N) -> %% Simple reservoir sampling sample_random(List, N, [], 0). sample_random([], _, Acc, _) -> Acc; sample_random(_, N, Acc, _) when length(Acc) >= N -> Acc; sample_random([H | T], N, Acc, Idx) -> %% Include with decreasing probability case rand:uniform() < (N / (Idx + 1)) of true when length(Acc) < N -> sample_random(T, N, [H | Acc], Idx + 1); true -> %% Replace random existing element ReplaceIdx = rand:uniform(length(Acc)), NewAcc = replace_nth(Acc, ReplaceIdx, H), sample_random(T, N, NewAcc, Idx + 1); false -> sample_random(T, N, Acc, Idx + 1) end. %% @private Replace Nth element in list. replace_nth([_ | T], 1, New) -> [New | T]; replace_nth([H | T], N, New) -> [H | replace_nth(T, N - 1, New)].