stouputils.mlflow.process_metrics_monitor module#

class ProcessMetricsMonitor(
pid: int | None = None,
children: bool = True,
sampling_interval: float = 10.0,
samples_before_logging: int = 1,
prefix: str = 'system/process/',
verbose: bool = False,
max_memory_megabytes: float | None = None,
max_cpu_count: float | None = None,
deadband: float = 0.01,
)[source]#

Bases: AbstractBothContextManager[ProcessMetricsMonitor]

Monitor that collects CPU, memory, I/O, and thread metrics for a specific process and (optionally) all its children, then logs them to MLflow.

This is the per-process counterpart of MLflow’s built-in log_system_metrics=True which only captures system-wide metrics. Here every metric is scoped to the process tree rooted at pid.

Metrics collected (all prefixed with system/process/):

  • cpu_usage_percentage - CPU % of the cores the process may use (see max_cpu_count)

  • memory_rss_megabytes - resident set size in MB

  • memory_vms_megabytes - virtual memory size in MB

  • memory_uss_megabytes - unique set size in MB (Linux only, falls back to RSS)

  • memory_usage_percentage - RSS as % of total available RAM (see max_memory_megabytes)

  • num_threads - total thread count across the tree

  • num_fds - total open file descriptors (Linux only, 0 on other OS)

  • io_read_megabytes - cumulative bytes read in MB (since process start)

  • io_write_megabytes - cumulative bytes written in MB (since process start)

Parameters:

  • pid – PID of the root process to monitor. Defaults to the current process (os.getpid()).

  • children – Whether to include child processes (recursively) in the metrics. Defaults to True.

  • sampling_interval – Seconds between each sample collection. Defaults to 10.

  • samples_before_logging – Number of samples to average before logging. Defaults to 1.

  • prefix – Metric name prefix. Defaults to "process/".

  • verbose – Whether to log verbose debug messages. Defaults to False.

  • max_memory_megabytes – Override the total memory in MB used to compute memory_usage_percentage. Defaults to MEMORY_MEGABYTES, the container’s own cap wherever cgroup v2 states one and the host’s total elsewhere.

  • max_cpu_count – Override the number of CPUs used to normalise cpu_usage_percentage. Defaults to cpu_limit(), read the same way. It is the raw ceiling rather than CPU_COUNT, which the thread-count environment variables override and which would then scale the percentage against a worker count.

  • deadband – Fraction of a curve’s own amplitude a sample may sit away from the line drawn without it, below which it is never written. 0.0 writes every sample.

Examples

> import mlflow
> from stouputils.mlflow.process_metrics_monitor import ProcessMetricsMonitor
> mlflow.set_experiment("my_experiment")
> with mlflow.start_run():
.     monitor = ProcessMetricsMonitor(pid=12345, children=True, sampling_interval=5)
.     monitor.start()
.     # ... do heavy work ...
.     monitor.finish()

Or as a context manager:

> import mlflow
> from stouputils.mlflow.process_metrics_monitor import ProcessMetricsMonitor
> mlflow.set_experiment("my_experiment")
> with mlflow.start_run(), ProcessMetricsMonitor(pid=12345):
.     # ... do heavy work ...
.     pass
pid: int[source]#

PID of the root process to monitor.

children: bool[source]#

Whether to include child processes recursively.

sampling_interval: float[source]#

Seconds between each sample collection.

samples_before_logging: int[source]#

Number of samples to average before logging.

prefix: str[source]#

Metric name prefix.

verbose: bool[source]#

Whether to log verbose debug messages.

max_memory_megabytes: float[source]#

Total memory in MB used as the denominator for memory_usage_percentage.

max_cpu_count: float[source]#

Number of CPUs used to normalise cpu_usage_percentage (psutil returns per-core %).

deadband_filter: DeadbandFilter[source]#

Thins each metric down to the samples the drawn curve cannot be read off without.

run_id: str | None[source]#

MLflow run ID captured at start time, ensures metrics are logged to the correct run from the daemon thread.

shutdown_event: Event[source]#

Event used to signal the monitoring thread to stop.

thread: Thread | None[source]#

Reference to the monitoring daemon thread.

_abc_impl = <_abc._abc_data object>[source]#
step: int[source]#

Current logging step counter.

processes: dict[int, Process][source]#

Persistent cache of monitored psutil.Process objects keyed by PID. Keeping the same objects across calls is required so that cpu_percent() has a non-zero interval to measure against (first call always returns 0).

start() → None[source]#

Start the background monitoring thread.

finish() → None[source]#

Stop monitoring

collect_once() → dict[str, float][source]#

Collect one snapshot of metrics for the process tree. :returns: A dictionary of metric names to values.

aggregate(
samples: list[dict[str, float]],
) → dict[str, float][source]#

Average the collected samples.

Parameters:

samples – List of metric dictionaries.

Returns:

A dictionary of averaged metric values.

publish(
metrics: dict[str, float],
) → None[source]#

Hand the aggregated metrics to the deadband filter and write whatever it releases.

The step still counts every sample, so a curve keeps the shape it was measured with however few rows it costs.

Parameters:

metrics – Aggregated metric values.

write(
batched: Mapping[int, Mapping[str, float]],
) → None[source]#

Send the released points to the active MLflow run, one round trip per step they belong to.

Parameters:

batched – Metric values keyed by the step each of them was measured at.

monitor_loop() → None[source]#

Main monitoring loop running in a daemon thread.