From d82d3192dce7971e81eb96a7dc226d412c604307 Mon Sep 17 00:00:00 2001 From: Aliaksei Kharlap Date: Sun, 9 Aug 2026 10:10:13 +0300 Subject: [PATCH] refactor system metrics plugin to use utils package --- influxdata/library/plugin_library.json | 4 +- influxdata/system_metrics/README.md | 75 +- influxdata/system_metrics/manifest.toml | 4 +- .../system_metrics/requirements-dev.txt | 3 + influxdata/system_metrics/requirements.txt | 1 + influxdata/system_metrics/system_metrics.py | 766 +++++++++++------ .../system_metrics/test_system_metrics.py | 788 ++++++++++++++++++ 7 files changed, 1338 insertions(+), 303 deletions(-) create mode 100644 influxdata/system_metrics/requirements-dev.txt create mode 100644 influxdata/system_metrics/test_system_metrics.py diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index 049e563..7370d99 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -149,8 +149,8 @@ "author": "InfluxData", "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/system_metrics/README.md", "required_plugins": [], - "required_libraries": ["psutil"], - "last_update": "2025-07-23", + "required_libraries": ["influxdata-plugin-utils>=0.3.0", "psutil"], + "last_update": "2026-08-06", "trigger_types_supported": ["scheduler"] }, { diff --git a/influxdata/system_metrics/README.md b/influxdata/system_metrics/README.md index a263772..2933458 100644 --- a/influxdata/system_metrics/README.md +++ b/influxdata/system_metrics/README.md @@ -18,24 +18,28 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor ### Optional parameters -| Parameter | Type | Default | Description | -|-------------------|---------|-------------|--------------------------------------------------------------------------------| -| `hostname` | string | `localhost` | Hostname to tag all metrics with for system identification | -| `include_cpu` | boolean | `true` | Include comprehensive CPU metrics collection (overall and per-core statistics) | -| `include_memory` | boolean | `true` | Include memory metrics collection (RAM usage, swap statistics, page faults) | -| `include_disk` | boolean | `true` | Include disk metrics collection (partition usage, I/O statistics, performance) | -| `include_network` | boolean | `true` | Include network metrics collection (interface statistics and error counts) | -| `max_retries` | integer | `3` | Maximum retry attempts on failure with graceful error handling | +| Parameter | Type | Default | Description | +|-------------------|---------|-------------|--------------------------------------------------------------------------------------------------| +| `hostname` | string | `localhost` | Hostname to tag all metrics with for system identification | +| `include_cpu` | boolean | `true` | Include comprehensive CPU metrics collection (overall and per-core statistics) | +| `include_memory` | boolean | `true` | Include memory metrics collection (RAM usage, swap statistics, page faults) | +| `include_disk` | boolean | `true` | Include disk metrics collection (partition usage, I/O statistics, performance) | +| `include_network` | boolean | `true` | Include network metrics collection (interface statistics and error counts) | +| `max_retries` | integer | `3` | Retry attempts per metric type; the group is skipped and the run continues once they are used up | *Note: This plugin has no required parameters. All parameters have sensible defaults.* +Boolean parameters accept `true`/`false`, `1`/`0`, `yes`/`no`, and `on`/`off`. A value the plugin cannot interpret is reported in the logs and the run collects nothing, so fix the trigger arguments and the next run recovers. + ### TOML configuration | Parameter | Type | Default | Description | |--------------------|--------|---------|----------------------------------------------------------------------------------| | `config_file_path` | string | none | TOML config file path relative to `PLUGIN_DIR` (required for TOML configuration) | -*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. +*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. Relative paths are resolved against the first directory that is set: `PLUGIN_DIR`, then `INFLUXDB3_PLUGIN_DIR`, then the parent of `VIRTUAL_ENV`. Only that directory is used — the file is not looked up in the remaining ones. + +Values in the TOML file override the inline trigger arguments. If the file cannot be read, the plugin logs an error and collects metrics using the inline arguments and defaults. #### Example TOML configuration @@ -46,8 +50,7 @@ For more information on using TOML configuration files, see the Using TOML Confi ## Software Requirements - **InfluxDB 3 Core/Enterprise**: with the Processing Engine enabled. -- **Python packages**: - - `psutil` (for system metrics collection) +- **Python packages**: `influxdata-plugin-utils>=0.3.0`, `psutil` ### Installation steps @@ -64,6 +67,7 @@ For more information on using TOML configuration files, see the Using TOML Confi 2. Install required Python packages: ```bash + influxdb3 install package "influxdata-plugin-utils>=0.3.0" influxdb3 install package psutil ``` @@ -167,21 +171,19 @@ influxdb3 query \ #### `process_scheduled_call()` -The main entry point for scheduled triggers. Collects system metrics based on configuration and writes them to InfluxDB. +The main entry point for scheduled triggers. Loads the configuration, then runs each enabled collector and writes the lines it built. A collector is retried up to `max_retries` times, and its lines are written only once it completes. ```python -def process_scheduled_call(influxdb3_local, call_time, args): - # Parse configuration - config = parse_config(args) - - # Collect metrics based on configuration - if config['include_cpu']: - collect_cpu_metrics(influxdb3_local, config['hostname']) - - if config['include_memory']: - collect_memory_metrics(influxdb3_local, config['hostname']) - - # ... additional metric collections +def process_scheduled_call(influxdb3_local, call_time, args=None): + config = _load_config(influxdb3_local, args, task_id) + + for config_key, metric_type, collect in _COLLECTORS: + if not config[config_key]: + continue + lines = _collect_with_retry( + influxdb3_local, collect, metric_type, hostname, max_retries, task_id + ) + write_data(influxdb3_local, lines, batch=False, retries=0) ``` ### Measurements and Fields @@ -193,6 +195,8 @@ Overall CPU statistics and metrics: - **Tags**: `host`, `cpu=total` - **Fields**: `user`, `system`, `idle`, `iowait`, `nice`, `irq`, `softirq`, `steal`, `guest`, `guest_nice`, `frequency_current`, `frequency_min`, `frequency_max`, `ctx_switches`, `interrupts`, `soft_interrupts`, `syscalls`, `load1`, `load5`, `load15` +The state shares (`user` through `guest_nice`) are derived from the change in the CPU time counters between two consecutive runs, so each value covers the interval between the previous run and the current one. They are absent on the first run after the trigger is created or restarted; the remaining fields are written from the first run on. + #### system_cpu_cores Per-core CPU statistics: @@ -200,6 +204,8 @@ Per-core CPU statistics: - **Tags**: `host`, `core` (core number) - **Fields**: `usage`, `user`, `system`, `idle`, `iowait`, `nice`, `irq`, `softirq`, `steal`, `guest`, `guest_nice`, `frequency_current`, `frequency_min`, `frequency_max` +Shares are derived the same way as in `system_cpu`; `usage` is the busy share of the core, everything except `idle` and `iowait`. + #### system_memory System memory statistics: @@ -237,11 +243,13 @@ Disk I/O statistics: #### system_disk_performance -Calculated disk performance metrics: +Disk performance rates, derived from the change in the I/O counters between two consecutive runs of the plugin: - **Tags**: `host`, `device` - **Fields**: `read_bytes_per_sec`, `write_bytes_per_sec`, `read_iops`, `write_iops`, `avg_read_latency_ms`, `avg_write_latency_ms`, `util_percent` +Each value covers the interval between the previous run and the current one, so the shorter the trigger interval, the finer the resolution. A device gets no line when there is nothing to compare against: on the first run after the trigger is created or restarted, and when its counters were reset (for example after the device was re-attached). + #### system_network Network interface statistics: @@ -257,14 +265,27 @@ Network interface statistics: **Solution**: The plugin will continue collecting other metrics even if some require elevated permissions. Run InfluxDB with appropriate permissions if disk I/O metrics are required. -#### Issue: Missing psutil library +#### Issue: Missing Python packages -**Solution**: Install the psutil package: +**Solution**: Install the required packages: ```bash +influxdb3 install package "influxdata-plugin-utils>=0.3.0" influxdb3 install package psutil ``` +#### Issue: No `system_disk_performance` data, or CPU shares are missing + +**Solution**: Both are derived from two consecutive runs. Wait for the second run of the trigger; if the values stay missing, check the logs for `No previous disk I/O sample` and `No previous CPU sample`, which repeat when the cached counters are lost on every run. + +#### Issue: One metric group is missing while the others are written + +**Solution**: A collector that keeps failing is skipped so the rest of the run survives. Look for `Failed to collect metrics after N retries` in the logs, followed by `skipped after repeated failures`, which names every group left out of that run. + +#### Issue: No metrics at all and a configuration error in the logs + +**Solution**: An invalid parameter value stops the run before any collection. Look for `Failed to load configuration` in the logs, which names the offending value, and fix the trigger arguments. A TOML file that cannot be read is a separate case: it is logged as `Failed to apply config file` and collection continues with the inline arguments. + #### Issue: High CPU usage from plugin **Solution**: Increase the trigger interval (for example, from `every:10s` to `every:30s`). Disable unnecessary metric types. Reduce the number of disk partitions monitored. diff --git a/influxdata/system_metrics/manifest.toml b/influxdata/system_metrics/manifest.toml index 4bba8ef..c719b43 100644 --- a/influxdata/system_metrics/manifest.toml +++ b/influxdata/system_metrics/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.2" [plugin] name = "system_metrics" -version = "1.1.0" +version = "1.2.0" description = "Collects system-level metrics (CPU, memory, disk, and network) using psutil and writes them to InfluxDB. Designed to run on a schedule and provide observability into host-level performance." triggers = ["process_scheduled_call"] homepage = "https://www.influxdata.com/" @@ -16,4 +16,4 @@ exclude = [ [dependencies] database_version = ">=3.0.0" -python = ["psutil"] +python = ["influxdata-plugin-utils>=0.3.0", "psutil"] diff --git a/influxdata/system_metrics/requirements-dev.txt b/influxdata/system_metrics/requirements-dev.txt new file mode 100644 index 0000000..aa356ca --- /dev/null +++ b/influxdata/system_metrics/requirements-dev.txt @@ -0,0 +1,3 @@ +pytest +influxdata-plugin-utils>=0.3.0 +psutil \ No newline at end of file diff --git a/influxdata/system_metrics/requirements.txt b/influxdata/system_metrics/requirements.txt index 0b574b5..ca82a2a 100644 --- a/influxdata/system_metrics/requirements.txt +++ b/influxdata/system_metrics/requirements.txt @@ -1 +1,2 @@ +influxdata-plugin-utils>=0.3.0 psutil \ No newline at end of file diff --git a/influxdata/system_metrics/system_metrics.py b/influxdata/system_metrics/system_metrics.py index faa19cb..de3ab4d 100644 --- a/influxdata/system_metrics/system_metrics.py +++ b/influxdata/system_metrics/system_metrics.py @@ -16,14 +16,14 @@ }, { "name": "include_memory", - "example": "true", + "example": "true", "description": "Include memory metrics collection", "required": false }, { "name": "include_disk", "example": "true", - "description": "Include disk metrics collection", + "description": "Include disk metrics collection", "required": false }, { @@ -41,305 +41,527 @@ { "name": "config_file_path", "example": "system_metrics_config_scheduler.toml", - "description": "Path to configuration file from PLUGIN_DIR env var", + "description": "Path to a TOML configuration file, relative to the plugin directory", "required": false } ] } """ -import psutil +import time import uuid -import os -import tomllib -from pathlib import Path -def collect_cpu_metrics(influxdb3_local, hostname, task_id): - # Get CPU frequencies +import psutil +from influxdata_plugin_utils.config import Validator, load_plugin_config +from influxdata_plugin_utils.parsing import parse_bool, parse_int +from influxdata_plugin_utils.write import build_line_typed, write_data + +_VALIDATORS = [ + Validator("hostname", default="localhost", cast=str), + Validator("include_cpu", default=True, cast=parse_bool), + Validator("include_memory", default=True, cast=parse_bool), + Validator("include_disk", default=True, cast=parse_bool), + Validator("include_network", default=True, cast=parse_bool), + Validator("max_retries", default=3, cast=lambda raw: parse_int(raw, minimum=0)), +] + +# Cached psutil counters, used to derive rates and shares between two runs +_DISK_IO_STATE_KEY = "system_metrics:disk_io" +_DISK_IO_COUNTERS = ( + "read_count", + "write_count", + "read_bytes", + "write_bytes", + "read_time", + "write_time", + "busy_time", +) + +_CPU_TIMES_STATE_KEY = "system_metrics:cpu_times" +_CPU_TIME_FIELDS = ( + "user", + "system", + "idle", + "iowait", + "nice", + "irq", + "softirq", + "steal", + "guest", + "guest_nice", +) + + +def _load_config(influxdb3_local, args: dict, task_id: str) -> dict | None: + """ + Load the plugin configuration, applying defaults and type casts. + + Values from a TOML file referenced by 'config_file_path' override the inline + trigger arguments. A config file that cannot be read is reported and skipped, + so collection continues with the inline arguments. + + Args: + influxdb3_local: InfluxDB client instance. + args (dict): Runtime arguments of the trigger. + task_id (str): Unique task identifier. + + Returns: + dict | None: Config values keyed by lower-case name, or None if the + inline arguments themselves are invalid. + """ + args = args or {} + config_file_path = args.get("config_file_path") + if config_file_path and not str(config_file_path).endswith(".toml"): + influxdb3_local.error( + f"[{task_id}] Invalid config file format: expected a .toml file" + ) + config_file_path = None + + try: + loaded = load_plugin_config(args, validators=_VALIDATORS, source="args") + except Exception as e: + influxdb3_local.error(f"[{task_id}] Failed to load configuration: {e}") + return None + + if config_file_path: + try: + loaded = load_plugin_config(args, validators=_VALIDATORS, source="merge") + influxdb3_local.info( + f"[{task_id}] Loaded configuration from {config_file_path}" + ) + except Exception as e: + influxdb3_local.error( + f"[{task_id}] Failed to apply config file '{config_file_path}': {e}. " + f"Continuing with inline arguments" + ) + + return {key.lower(): value for key, value in loaded.as_dict().items()} + + +def _float_fields(**values) -> dict: + """Type every value as a float64 field.""" + return {name: (value, "float") for name, value in values.items()} + + +def _uint_fields(**values) -> dict: + """Type every value as a uint64 field.""" + return {name: (value, "uint") for name, value in values.items()} + + +def _cpu_times_sample(times) -> dict: + """Snapshot the cumulative CPU time counters of one CPU.""" + return {name: getattr(times, name, 0.0) for name in _CPU_TIME_FIELDS} + + +def _cpu_total_time(sample: dict) -> float: + # guest time is already counted inside user and nice, so it is excluded here + return sum(sample.values()) - sample["guest"] - sample["guest_nice"] + + +def _cpu_percent_fields(previous: dict, current: dict) -> dict | None: + """ + Derive the share of every CPU state from two cumulative samples. + + Returns None when the samples cannot be compared, which happens on the first + run of the plugin and after the counters are reset. + """ + total = _cpu_total_time(current) - _cpu_total_time(previous) + if total <= 0: + return None + + percentages = {} + for name in _CPU_TIME_FIELDS: + delta = current[name] - previous[name] + if delta < 0: + return None + percentages[name] = round(min(100.0, delta * 100.0 / total), 1) + return percentages + + +def _cpu_usage_percent(percentages: dict) -> float: + """Busy share of a CPU: everything except idle and waiting for I/O.""" + return round(max(0.0, 100.0 - percentages["idle"] - percentages["iowait"]), 1) + + +def collect_cpu_metrics(influxdb3_local, hostname: str, task_id: str) -> list: + """Build overall CPU lines plus one line per core.""" cpu_freq = psutil.cpu_freq(percpu=False) cpu_stats = psutil.cpu_stats() - cpu_times = psutil.cpu_times_percent() load_avg = psutil.getloadavg() - - # Overall CPU usage and stats - line = LineBuilder("system_cpu")\ - .tag("host", hostname)\ - .tag("cpu", "total")\ - .float64_field("user", cpu_times.user)\ - .float64_field("system", cpu_times.system)\ - .float64_field("idle", cpu_times.idle)\ - .float64_field("iowait", getattr(cpu_times, 'iowait', 0))\ - .float64_field("nice", getattr(cpu_times, 'nice', 0))\ - .float64_field("irq", getattr(cpu_times, 'irq', 0))\ - .float64_field("softirq", getattr(cpu_times, 'softirq', 0))\ - .float64_field("steal", getattr(cpu_times, 'steal', 0))\ - .float64_field("guest", getattr(cpu_times, 'guest', 0))\ - .float64_field("guest_nice", getattr(cpu_times, 'guest_nice', 0))\ - .float64_field("frequency_current", getattr(cpu_freq, 'current', 0))\ - .float64_field("frequency_min", getattr(cpu_freq, 'min', 0))\ - .float64_field("frequency_max", getattr(cpu_freq, 'max', 0))\ - .uint64_field("ctx_switches", cpu_stats.ctx_switches)\ - .uint64_field("interrupts", cpu_stats.interrupts)\ - .uint64_field("soft_interrupts", cpu_stats.soft_interrupts)\ - .uint64_field("syscalls", getattr(cpu_stats, 'syscalls', 0))\ - .float64_field("load1", load_avg[0])\ - .float64_field("load5", load_avg[1])\ - .float64_field("load15", load_avg[2]) - influxdb3_local.write(line) - - # Per CPU core metrics + + previous: dict = influxdb3_local.cache.get(_CPU_TIMES_STATE_KEY, default={}) + current: dict = { + "total": _cpu_times_sample(psutil.cpu_times()), + "per_cpu": [_cpu_times_sample(times) for times in psutil.cpu_times(percpu=True)], + } + influxdb3_local.cache.put(_CPU_TIMES_STATE_KEY, current, None) + + if not previous: + influxdb3_local.info( + f"[{task_id}] No previous CPU sample, " + f"usage percentages start with the next run" + ) + + typed_fields = { + **_float_fields( + frequency_current=getattr(cpu_freq, "current", 0), + frequency_min=getattr(cpu_freq, "min", 0), + frequency_max=getattr(cpu_freq, "max", 0), + load1=load_avg[0], + load5=load_avg[1], + load15=load_avg[2], + ), + **_uint_fields( + ctx_switches=cpu_stats.ctx_switches, + interrupts=cpu_stats.interrupts, + soft_interrupts=cpu_stats.soft_interrupts, + syscalls=getattr(cpu_stats, "syscalls", 0), + ), + } + if previous: + total_percentages = _cpu_percent_fields(previous["total"], current["total"]) + if total_percentages: + typed_fields.update(_float_fields(**total_percentages)) + + lines = [ + build_line_typed( + LineBuilder, + "system_cpu", + tags={"host": hostname, "cpu": "total"}, + typed_fields=typed_fields, + ) + ] + try: - per_cpu_percent = psutil.cpu_percent(interval=None, percpu=True) - per_cpu_times = psutil.cpu_times_percent(percpu=True) per_cpu_freq = psutil.cpu_freq(percpu=True) - - for core_id in range(len(per_cpu_percent)): - line = LineBuilder("system_cpu_cores")\ - .tag("host", hostname)\ - .tag("core", str(core_id)) - - # Add usage percentage - line.float64_field("usage", per_cpu_percent[core_id]) - - # Add CPU time breakdowns if available - if core_id < len(per_cpu_times): - core_times = per_cpu_times[core_id] - line.float64_field("user", core_times.user)\ - .float64_field("system", core_times.system)\ - .float64_field("idle", core_times.idle)\ - .float64_field("iowait", getattr(core_times, 'iowait', 0))\ - .float64_field("nice", getattr(core_times, 'nice', 0))\ - .float64_field("irq", getattr(core_times, 'irq', 0))\ - .float64_field("softirq", getattr(core_times, 'softirq', 0))\ - .float64_field("steal", getattr(core_times, 'steal', 0))\ - .float64_field("guest", getattr(core_times, 'guest', 0))\ - .float64_field("guest_nice", getattr(core_times, 'guest_nice', 0)) - - # Add frequency metrics if available - if per_cpu_freq and core_id < len(per_cpu_freq): - freq = per_cpu_freq[core_id] - line.float64_field("frequency_current", freq.current)\ - .float64_field("frequency_min", getattr(freq, 'min', 0))\ - .float64_field("frequency_max", getattr(freq, 'max', 0)) - - influxdb3_local.write(line) except Exception as e: - influxdb3_local.warn(f"[{task_id}] Error collecting per-core CPU metrics: {str(e)}") + per_cpu_freq = [] + influxdb3_local.warn(f"[{task_id}] Error reading per-core CPU frequency: {e}") + + previous_per_cpu: list = previous.get("per_cpu", []) + for core_id, core_times in enumerate(current["per_cpu"]): + core_fields = {} + + if core_id < len(previous_per_cpu): + core_percentages = _cpu_percent_fields(previous_per_cpu[core_id], core_times) + if core_percentages: + core_fields.update(_float_fields(**core_percentages)) + core_fields.update( + _float_fields(usage=_cpu_usage_percent(core_percentages)) + ) + + if per_cpu_freq and core_id < len(per_cpu_freq): + freq = per_cpu_freq[core_id] + core_fields.update( + _float_fields( + frequency_current=freq.current, + frequency_min=getattr(freq, "min", 0), + frequency_max=getattr(freq, "max", 0), + ) + ) + + if not core_fields: + continue - -def collect_memory_metrics(influxdb3_local, hostname, task_id): - # Virtual memory metrics + lines.append( + build_line_typed( + LineBuilder, + "system_cpu_cores", + tags={"host": hostname, "core": str(core_id)}, + typed_fields=core_fields, + ) + ) + + return lines + + +def collect_memory_metrics(influxdb3_local, hostname: str, task_id: str) -> list: + """Build memory, swap, and page fault lines.""" mem = psutil.virtual_memory() swap = psutil.swap_memory() - - # Main memory metrics - line = LineBuilder("system_memory")\ - .tag("host", hostname)\ - .uint64_field("total", mem.total)\ - .uint64_field("available", mem.available)\ - .uint64_field("used", mem.used)\ - .uint64_field("free", mem.free)\ - .uint64_field("active", getattr(mem, 'active', 0))\ - .uint64_field("inactive", getattr(mem, 'inactive', 0))\ - .uint64_field("buffers", getattr(mem, 'buffers', 0))\ - .uint64_field("cached", getattr(mem, 'cached', 0))\ - .uint64_field("shared", getattr(mem, 'shared', 0))\ - .uint64_field("slab", getattr(mem, 'slab', 0))\ - .float64_field("percent", mem.percent) - influxdb3_local.write(line) - - # Swap metrics in separate measurement - line = LineBuilder("system_swap")\ - .tag("host", hostname)\ - .uint64_field("total", swap.total)\ - .uint64_field("used", swap.used)\ - .uint64_field("free", swap.free)\ - .float64_field("percent", swap.percent)\ - .uint64_field("sin", swap.sin)\ - .uint64_field("sout", swap.sout) - influxdb3_local.write(line) - - # Try to collect memory page faults if available + + lines = [ + build_line_typed( + LineBuilder, + "system_memory", + tags={"host": hostname}, + typed_fields={ + **_uint_fields( + total=mem.total, + available=mem.available, + used=mem.used, + free=mem.free, + active=getattr(mem, "active", 0), + inactive=getattr(mem, "inactive", 0), + buffers=getattr(mem, "buffers", 0), + cached=getattr(mem, "cached", 0), + shared=getattr(mem, "shared", 0), + slab=getattr(mem, "slab", 0), + ), + **_float_fields(percent=mem.percent), + }, + ), + build_line_typed( + LineBuilder, + "system_swap", + tags={"host": hostname}, + typed_fields={ + **_uint_fields( + total=swap.total, + used=swap.used, + free=swap.free, + sin=swap.sin, + sout=swap.sout, + ), + **_float_fields(percent=swap.percent), + }, + ), + ] + try: page_faults = psutil.Process().memory_full_info() - line = LineBuilder("system_memory_faults")\ - .tag("host", hostname)\ - .uint64_field("page_faults", getattr(page_faults, 'num_page_faults', 0))\ - .uint64_field("major_faults", getattr(page_faults, 'maj_faults', 0))\ - .uint64_field("minor_faults", getattr(page_faults, 'min_faults', 0))\ - .uint64_field("rss", getattr(page_faults, 'rss', 0))\ - .uint64_field("vms", getattr(page_faults, 'vms', 0))\ - .uint64_field("dirty", getattr(page_faults, 'dirty', 0))\ - .uint64_field("uss", getattr(page_faults, 'uss', 0))\ - .uint64_field("pss", getattr(page_faults, 'pss', 0)) - influxdb3_local.write(line) - except (psutil.AccessDenied, psutil.Error): + lines.append( + build_line_typed( + LineBuilder, + "system_memory_faults", + tags={"host": hostname}, + typed_fields=_uint_fields( + page_faults=getattr(page_faults, "num_page_faults", 0), + major_faults=getattr(page_faults, "maj_faults", 0), + minor_faults=getattr(page_faults, "min_faults", 0), + rss=getattr(page_faults, "rss", 0), + vms=getattr(page_faults, "vms", 0), + dirty=getattr(page_faults, "dirty", 0), + uss=getattr(page_faults, "uss", 0), + pss=getattr(page_faults, "pss", 0), + ), + ) + ) + except psutil.Error: pass -def collect_disk_metrics(influxdb3_local, hostname, task_id): - # Collect disk partition usage metrics + return lines + + +def _disk_io_sample(stats, timestamp_ns: int) -> dict: + """Snapshot the cumulative I/O counters of one device.""" + sample = {name: getattr(stats, name, 0) for name in _DISK_IO_COUNTERS} + sample["timestamp_ns"] = timestamp_ns + return sample + + +def _disk_performance_fields(previous: dict, current: dict) -> dict | None: + """ + Derive I/O rates from two counter samples of the same device. + + Returns None when the samples cannot be compared, which happens on the first + run of the plugin and after the counters are reset. + """ + elapsed_seconds = (current["timestamp_ns"] - previous["timestamp_ns"]) / 1_000_000_000 + if elapsed_seconds <= 0: + return None + + deltas = {name: current[name] - previous[name] for name in _DISK_IO_COUNTERS} + if any(delta < 0 for delta in deltas.values()): + return None + + return _float_fields( + read_bytes_per_sec=deltas["read_bytes"] / elapsed_seconds, + write_bytes_per_sec=deltas["write_bytes"] / elapsed_seconds, + read_iops=deltas["read_count"] / elapsed_seconds, + write_iops=deltas["write_count"] / elapsed_seconds, + avg_read_latency_ms=( + deltas["read_time"] / deltas["read_count"] if deltas["read_count"] else 0 + ), + avg_write_latency_ms=( + deltas["write_time"] / deltas["write_count"] if deltas["write_count"] else 0 + ), + util_percent=deltas["busy_time"] / (elapsed_seconds * 1000) * 100, + ) + + +def collect_disk_metrics(influxdb3_local, hostname: str, task_id: str) -> list: + """Build per-partition usage lines plus per-device I/O and rate lines.""" + lines = [] + for partition in psutil.disk_partitions(all=False): try: usage = psutil.disk_usage(partition.mountpoint) - line = LineBuilder("system_disk_usage")\ - .tag("host", hostname)\ - .tag("device", partition.device)\ - .tag("mountpoint", partition.mountpoint)\ - .tag("fstype", partition.fstype)\ - .uint64_field("total", usage.total)\ - .uint64_field("used", usage.used)\ - .uint64_field("free", usage.free)\ - .float64_field("percent", usage.percent) - influxdb3_local.write(line) - except PermissionError: + except OSError: + # an unreadable or disconnected mountpoint must not abort the collector continue + lines.append( + build_line_typed( + LineBuilder, + "system_disk_usage", + tags={ + "host": hostname, + "device": partition.device, + "mountpoint": partition.mountpoint, + "fstype": partition.fstype, + }, + typed_fields={ + **_uint_fields(total=usage.total, used=usage.used, free=usage.free), + **_float_fields(percent=usage.percent), + }, + ) + ) - # Collect disk I/O statistics try: disk_io = psutil.disk_io_counters(perdisk=True) - for disk_name, stats in disk_io.items(): - line = LineBuilder("system_disk_io")\ - .tag("host", hostname)\ - .tag("device", disk_name)\ - .uint64_field("reads", stats.read_count)\ - .uint64_field("writes", stats.write_count)\ - .uint64_field("read_bytes", stats.read_bytes)\ - .uint64_field("write_bytes", stats.write_bytes)\ - .uint64_field("read_time", stats.read_time)\ - .uint64_field("write_time", stats.write_time)\ - .uint64_field("busy_time", getattr(stats, 'busy_time', 0))\ - .uint64_field("read_merged_count", getattr(stats, 'read_merged_count', 0))\ - .uint64_field("write_merged_count", getattr(stats, 'write_merged_count', 0)) - influxdb3_local.write(line) - - # Calculate and write IOPS and throughput metrics - # Note: These are instantaneous rates since the last measurement - line = LineBuilder("system_disk_performance")\ - .tag("host", hostname)\ - .tag("device", disk_name)\ - .float64_field("read_bytes_per_sec", getattr(stats, 'read_bytes_per_sec', 0))\ - .float64_field("write_bytes_per_sec", getattr(stats, 'write_bytes_per_sec', 0))\ - .float64_field("read_iops", getattr(stats, 'read_count_per_sec', 0))\ - .float64_field("write_iops", getattr(stats, 'write_count_per_sec', 0))\ - .float64_field("avg_read_latency_ms", stats.read_time / stats.read_count if stats.read_count > 0 else 0)\ - .float64_field("avg_write_latency_ms", stats.write_time / stats.write_count if stats.write_count > 0 else 0)\ - .float64_field("util_percent", getattr(stats, 'busy_time_percent', 0)) - influxdb3_local.write(line) except (psutil.Error, AttributeError) as e: - influxdb3_local.warn(f"[{task_id}] Error collecting disk I/O metrics: {str(e)}") - - -def collect_network_metrics(influxdb3_local, hostname, task_id): - net_io = psutil.net_io_counters(pernic=True) - - for interface, stats in net_io.items(): - line = LineBuilder("system_network")\ - .tag("host", hostname)\ - .tag("interface", interface)\ - .uint64_field("bytes_sent", stats.bytes_sent)\ - .uint64_field("bytes_recv", stats.bytes_recv)\ - .uint64_field("packets_sent", stats.packets_sent)\ - .uint64_field("packets_recv", stats.packets_recv)\ - .uint64_field("errin", stats.errin)\ - .uint64_field("errout", stats.errout)\ - .uint64_field("dropin", stats.dropin)\ - .uint64_field("dropout", stats.dropout) - influxdb3_local.write(line) - -def process_scheduled_call(influxdb3_local, time, args=None): - task_id = str(uuid.uuid4()) - - try: - # Load configuration from TOML file if provided - if args and 'config_file_path' in args: - config_file = args['config_file_path'] - if not config_file.endswith('.toml'): + influxdb3_local.warn(f"[{task_id}] Error collecting disk I/O metrics: {e}") + return lines + + timestamp_ns: int = time.time_ns() + previous_samples: dict = influxdb3_local.cache.get(_DISK_IO_STATE_KEY, default={}) + current_samples: dict = {} + + if not previous_samples: + influxdb3_local.info( + f"[{task_id}] No previous disk I/O sample, " + f"performance rates start with the next run" + ) + + for device, stats in disk_io.items(): + lines.append( + build_line_typed( + LineBuilder, + "system_disk_io", + tags={"host": hostname, "device": device}, + typed_fields=_uint_fields( + reads=stats.read_count, + writes=stats.write_count, + read_bytes=stats.read_bytes, + write_bytes=stats.write_bytes, + read_time=stats.read_time, + write_time=stats.write_time, + busy_time=getattr(stats, "busy_time", 0), + read_merged_count=getattr(stats, "read_merged_count", 0), + write_merged_count=getattr(stats, "write_merged_count", 0), + ), + ) + ) + + current_samples[device] = _disk_io_sample(stats, timestamp_ns) + previous_sample = previous_samples.get(device) + if previous_sample is None: + continue + + performance = _disk_performance_fields(previous_sample, current_samples[device]) + if performance is None: + continue + + lines.append( + build_line_typed( + LineBuilder, + "system_disk_performance", + tags={"host": hostname, "device": device}, + typed_fields=performance, + ) + ) + + influxdb3_local.cache.put(_DISK_IO_STATE_KEY, current_samples, ttl=None) + + return lines + + +def collect_network_metrics(influxdb3_local, hostname: str, task_id: str) -> list: + """Build one line per network interface.""" + return [ + build_line_typed( + LineBuilder, + "system_network", + tags={"host": hostname, "interface": interface}, + typed_fields=_uint_fields( + bytes_sent=stats.bytes_sent, + bytes_recv=stats.bytes_recv, + packets_sent=stats.packets_sent, + packets_recv=stats.packets_recv, + errin=stats.errin, + errout=stats.errout, + dropin=stats.dropin, + dropout=stats.dropout, + ), + ) + for interface, stats in psutil.net_io_counters(pernic=True).items() + ] + + +_COLLECTORS = ( + ("include_cpu", "CPU", collect_cpu_metrics), + ("include_memory", "memory", collect_memory_metrics), + ("include_disk", "disk", collect_disk_metrics), + ("include_network", "network", collect_network_metrics), +) + + +def _collect_with_retry( + influxdb3_local, + collect, + metric_type: str, + hostname: str, + max_retries: int, + task_id: str, +) -> list | None: + """ + Run one collector, retrying on failure. + + Returns the lines it built, or None once the retries are used up, so that the + remaining collectors still run and their points reach the database. + """ + for attempt in range(max_retries + 1): + try: + return collect(influxdb3_local, hostname, task_id) + except Exception as e: + if attempt == max_retries: influxdb3_local.error( - f"[{task_id}] Invalid config file format: expected a .toml file" + f"[{task_id}] Failed to collect {metric_type} metrics " + f"after {max_retries} retries: {e}" ) - else: - plugin_dir = os.environ.get('PLUGIN_DIR') - if plugin_dir: - config_path = os.path.join(plugin_dir, config_file) - else: - # Fallbacks for servers where the operator has not exported PLUGIN_DIR: - # - INFLUXDB3_PLUGIN_DIR: set when the server is configured via env var - # - VIRTUAL_ENV: exported by the processing engine; default venv is /.venv - candidates = [] - if influxdb3_plugin_dir := os.environ.get('INFLUXDB3_PLUGIN_DIR'): - candidates.append(influxdb3_plugin_dir) - if virtual_env := os.environ.get('VIRTUAL_ENV'): - candidates.append(str(Path(virtual_env).parent)) - - resolved = None - for base in candidates: - candidate = os.path.join(base, config_file) - if os.path.exists(candidate): - resolved = candidate - break - - if resolved: - config_path = resolved - else: - # Match original behavior: continue with inline args - # when the config file cannot be resolved. - config_path = None - candidate_str = ', '.join(candidates) if candidates else 'none available' - influxdb3_local.error( - f"[{task_id}] PLUGIN_DIR env var not set and config file path '{config_file}' was not found via fallbacks (tried: {candidate_str})" - ) - - if config_path: - try: - with open(config_path, 'rb') as f: - config = tomllib.load(f) - # Override args with config values - args.update(config) - influxdb3_local.info(f"[{task_id}] Loaded configuration from {config_path}") - except Exception: - influxdb3_local.error(f"[{task_id}] Failed to read config file") - - # Set default values if args is None - if args is None: - args = {} - - # Get configuration values with defaults - hostname = args.get("hostname", "localhost") - include_cpu = str(args.get("include_cpu", "true")).lower() == "true" - include_memory = str(args.get("include_memory", "true")).lower() == "true" - include_disk = str(args.get("include_disk", "true")).lower() == "true" - include_network = str(args.get("include_network", "true")).lower() == "true" - max_retries = int(args.get("max_retries", 3)) - - influxdb3_local.info(f"[{task_id}] Starting system metrics collection for host: {hostname}") - - # Collect metrics with retry logic - def collect_with_retry(collect_func, metric_type): - for attempt in range(max_retries + 1): - try: - collect_func(influxdb3_local, hostname, task_id) - break - except Exception as e: - if attempt == max_retries: - influxdb3_local.error(f"[{task_id}] Failed to collect {metric_type} metrics after {max_retries} retries: {e}") - raise - influxdb3_local.warn(f"[{task_id}] {metric_type} metrics collection attempt {attempt + 1} failed, retrying: {e}") - - # Collect enabled metrics - if include_cpu: - collect_with_retry(collect_cpu_metrics, "CPU") - - if include_memory: - collect_with_retry(collect_memory_metrics, "memory") - - if include_disk: - collect_with_retry(collect_disk_metrics, "disk") - - if include_network: - collect_with_retry(collect_network_metrics, "network") - - influxdb3_local.info(f"[{task_id}] Successfully collected system metrics for host: {hostname}") - + return None + influxdb3_local.warn( + f"[{task_id}] {metric_type} metrics collection attempt " + f"{attempt + 1} failed, retrying: {e}" + ) + + +def process_scheduled_call(influxdb3_local, call_time, args=None): + task_id = str(uuid.uuid4()) + + config: dict | None = _load_config(influxdb3_local, args, task_id) + if config is None: + return + + hostname: str = config["hostname"] + max_retries: int = config["max_retries"] + + influxdb3_local.info( + f"[{task_id}] Starting system metrics collection for host: {hostname}" + ) + + skipped: list[str] = [] + try: + for config_key, metric_type, collect in _COLLECTORS: + if not config[config_key]: + continue + lines = _collect_with_retry( + influxdb3_local, collect, metric_type, hostname, max_retries, task_id + ) + if lines is None: + skipped.append(metric_type) + continue + write_data(influxdb3_local, lines, batch=False, retries=0) except Exception as e: - influxdb3_local.error(f"[{task_id}] Error collecting system metrics: {str(e)}") - raise \ No newline at end of file + influxdb3_local.error(f"[{task_id}] Error collecting system metrics: {e}") + raise + + if skipped: + influxdb3_local.error( + f"[{task_id}] Collected system metrics for host: {hostname}, " + f"skipped after repeated failures: {', '.join(skipped)}" + ) + else: + influxdb3_local.info( + f"[{task_id}] Successfully collected system metrics for host: {hostname}" + ) \ No newline at end of file diff --git a/influxdata/system_metrics/test_system_metrics.py b/influxdata/system_metrics/test_system_metrics.py new file mode 100644 index 0000000..9c559e1 --- /dev/null +++ b/influxdata/system_metrics/test_system_metrics.py @@ -0,0 +1,788 @@ +from collections import OrderedDict, namedtuple +from typing import Optional + +import psutil +import pytest + +class InfluxDBError(Exception): + pass + + +class InvalidMeasurementError(InfluxDBError): + pass + + +class InvalidKeyError(InfluxDBError): + pass + + +class InvalidLineError(InfluxDBError): + pass + + +class LineBuilder: + def __init__(self, measurement: str): + if " " in measurement: + raise InvalidMeasurementError("Measurement name cannot contain spaces") + self.measurement = measurement + self.tags: OrderedDict[str, str] = OrderedDict() + self.fields: OrderedDict[str, str] = OrderedDict() + self._timestamp_ns: Optional[int] = None + + def _validate_key(self, key: str, key_type: str) -> None: + if not key: + raise InvalidKeyError(f"{key_type} key cannot be empty") + if " " in key: + raise InvalidKeyError(f"{key_type} key '{key}' cannot contain spaces") + if "," in key: + raise InvalidKeyError(f"{key_type} key '{key}' cannot contain commas") + if "=" in key: + raise InvalidKeyError(f"{key_type} key '{key}' cannot contain equals signs") + + def tag(self, key: str, value: str) -> "LineBuilder": + self._validate_key(key, "tag") + self.tags[key] = str(value) + return self + + def uint64_field(self, key: str, value: int) -> "LineBuilder": + self._validate_key(key, "field") + if value < 0: + raise ValueError(f"uint64 field '{key}' cannot be negative") + self.fields[key] = f"{value}u" + return self + + def int64_field(self, key: str, value: int) -> "LineBuilder": + self._validate_key(key, "field") + self.fields[key] = f"{value}i" + return self + + def float64_field(self, key: str, value: float) -> "LineBuilder": + self._validate_key(key, "field") + self.fields[key] = f"{int(value)}.0" if value % 1 == 0 else str(value) + return self + + def string_field(self, key: str, value: str) -> "LineBuilder": + self._validate_key(key, "field") + escaped_value = value.replace("\\", "\\\\").replace('"', '\\"') + self.fields[key] = f'"{escaped_value}"' + return self + + def bool_field(self, key: str, value: bool) -> "LineBuilder": + self._validate_key(key, "field") + self.fields[key] = "t" if value else "f" + return self + + def time_ns(self, timestamp_ns: int) -> "LineBuilder": + self._timestamp_ns = timestamp_ns + return self + + def build(self) -> str: + line = self.measurement.replace(",", "\\,").replace(" ", "\\ ") + if self.tags: + line += "," + ",".join(f"{k}={v}" for k, v in self.tags.items()) + if not self.fields: + raise InvalidLineError(f"At least one field is required: {line}") + line += " " + ",".join(f"{k}={v}" for k, v in self.fields.items()) + if self._timestamp_ns is not None: + line += f" {self._timestamp_ns}" + return line + + +class FakeCache: + def __init__(self): + self._values = {} + self.ttls = {} + + def get(self, key, default=None, use_global=None): + return self._values.get(key, default) + + def put(self, key, value, ttl=None, use_global=None): + self._values[key] = value + self.ttls[key] = ttl + + def delete(self, key, use_global=None): + return self._values.pop(key, None) is not None + + +class FakeInfluxdb3Local: + def __init__(self): + self.cache = FakeCache() + self.logs = [] + self.writes = [] + + def info(self, message): + self.logs.append(("info", message)) + + def warn(self, message): + self.logs.append(("warn", message)) + + def error(self, message): + self.logs.append(("error", message)) + + def write(self, line): + self.writes.append(line.build()) + + def write_sync(self, line, no_sync=False): + raise AssertionError("plugin must use the buffered write API") + + def messages(self, level): + return [message for log_level, message in self.logs if log_level == level] + + +import system_metrics +from system_metrics import ( + _COLLECTORS, + _CPU_TIME_FIELDS, + _CPU_TIMES_STATE_KEY, + _DISK_IO_STATE_KEY, + _collect_with_retry, + _cpu_percent_fields, + _cpu_usage_percent, + _disk_io_sample, + _disk_performance_fields, + _float_fields, + _load_config, + _uint_fields, + collect_cpu_metrics, + collect_disk_metrics, + collect_memory_metrics, + collect_network_metrics, + process_scheduled_call, +) + +FakePartition = namedtuple("FakePartition", "device mountpoint fstype") +FakeUsage = namedtuple("FakeUsage", "total used free percent") +FakeNetIO = namedtuple( + "FakeNetIO", + "bytes_sent bytes_recv packets_sent packets_recv errin errout dropin dropout", +) + + +class FakeDiskIO: + """Stand-in for psutil's sdiskio, minus the counters a platform may not expose.""" + + def __init__(self, **counters): + self.__dict__.update(counters) + + +@pytest.fixture +def client(): + return FakeInfluxdb3Local() + + +@pytest.fixture(autouse=True) +def line_builder(monkeypatch): + """The engine injects LineBuilder as a global; tests use the vendored copy.""" + monkeypatch.setattr(system_metrics, "LineBuilder", LineBuilder, raising=False) + + +@pytest.fixture +def plugin_dir(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + return tmp_path + + +def raiser(error): + """Build a stand-in that raises instead of returning psutil data.""" + + def raise_error(*args, **kwargs): + raise error + + return raise_error + + +def parse_line(text): + """Split a built line into (measurement, tags, fields).""" + head, field_part = text.split(" ", 1) + parts = head.split(",") + tags = dict(item.split("=", 1) for item in parts[1:]) + fields = dict(item.split("=", 1) for item in field_part.split(",")) + return parts[0], tags, fields + + +def measurements(lines): + return [parse_line(line.build())[0] for line in lines] + + +# -------------------------------------------------------------------------- +# Configuration +# -------------------------------------------------------------------------- + + +def test_config_defaults(client): + config = _load_config(client, None, "task") + + assert config["hostname"] == "localhost" + assert config["max_retries"] == 3 + assert all( + config[key] + for key in ("include_cpu", "include_memory", "include_disk", "include_network") + ) + + +def test_config_casts_inline_args(client): + config = _load_config( + client, + { + "hostname": "web-1", + "include_cpu": "no", + "include_memory": "0", + "include_disk": "off", + "include_network": "yes", + "max_retries": "5", + }, + "task", + ) + + assert config["hostname"] == "web-1" + assert config["include_cpu"] is False + assert config["include_memory"] is False + assert config["include_disk"] is False + assert config["include_network"] is True + assert config["max_retries"] == 5 + + +@pytest.mark.parametrize( + "args, expected", + [ + ({"include_cpu": "maybe"}, "Invalid boolean: 'maybe'"), + ({"max_retries": "abc"}, "Invalid integer: 'abc'"), + ({"max_retries": "-1"}, "below minimum 0"), + ], +) +def test_config_rejects_invalid_values(client, args, expected): + assert _load_config(client, args, "task") is None + assert expected in client.messages("error")[0] + + +def test_config_toml_overrides_inline_args(client, plugin_dir): + (plugin_dir / "sm.toml").write_text( + 'hostname = "from-toml"\ninclude_cpu = false\nmax_retries = 1\n' + ) + + config = _load_config( + client, + {"hostname": "from-args", "config_file_path": "sm.toml"}, + "task", + ) + + assert config["hostname"] == "from-toml" + assert config["include_cpu"] is False + assert config["max_retries"] == 1 + assert config["include_memory"] is True # untouched default + assert "Loaded configuration from sm.toml" in client.messages("info")[0] + + +def test_config_missing_toml_keeps_inline_args(client, plugin_dir): + config = _load_config( + client, + {"hostname": "fallback", "config_file_path": "absent.toml"}, + "task", + ) + + assert config["hostname"] == "fallback" + error = client.messages("error")[0] + assert "Failed to apply config file 'absent.toml'" in error + assert "Continuing with inline arguments" in error + + +def test_config_ignores_file_with_wrong_extension(client, plugin_dir): + (plugin_dir / "sm.yaml").write_text('hostname = "from-file"\n') + + config = _load_config( + client, + {"hostname": "from-args", "config_file_path": "sm.yaml"}, + "task", + ) + + assert config["hostname"] == "from-args" + assert "expected a .toml file" in client.messages("error")[0] + + +def test_config_invalid_toml_value_keeps_inline_args(client, plugin_dir): + (plugin_dir / "sm.toml").write_text('include_cpu = "maybe"\n') + + config = _load_config( + client, + {"include_cpu": "false", "config_file_path": "sm.toml"}, + "task", + ) + + assert config["include_cpu"] is False + assert "Invalid boolean: 'maybe'" in client.messages("error")[0] + + +def test_config_accepts_absolute_toml_path(client, tmp_path, monkeypatch): + monkeypatch.delenv("PLUGIN_DIR", raising=False) + config_file = tmp_path / "abs.toml" + config_file.write_text('hostname = "abs-path"\n') + + config = _load_config(client, {"config_file_path": str(config_file)}, "task") + + assert config["hostname"] == "abs-path" + + +# -------------------------------------------------------------------------- +# Field typing helpers +# -------------------------------------------------------------------------- + + +def test_field_helpers_tag_values_with_their_type(): + assert _float_fields(percent=1.5) == {"percent": (1.5, "float")} + assert _uint_fields(total=10, used=4) == { + "total": (10, "uint"), + "used": (4, "uint"), + } + + +# -------------------------------------------------------------------------- +# CPU shares +# -------------------------------------------------------------------------- + + +def _cpu_sample(**counters): + sample = {name: 0.0 for name in _CPU_TIME_FIELDS} + sample.update(counters) + return sample + + +def test_cpu_percent_fields_computes_shares_of_the_interval(): + previous = _cpu_sample(user=100.0, system=50.0, idle=850.0) + current = _cpu_sample(user=140.0, system=60.0, idle=1000.0) + + percentages = _cpu_percent_fields(previous, current) + + assert percentages["user"] == 20.0 + assert percentages["system"] == 5.0 + assert percentages["idle"] == 75.0 + assert _cpu_usage_percent(percentages) == 25.0 + + +def test_cpu_percent_fields_excludes_guest_time_from_the_total(): + current = _cpu_sample(user=50.0, idle=50.0, guest=50.0) + + assert _cpu_percent_fields(_cpu_sample(), current)["user"] == 50.0 + + +def test_cpu_percent_fields_requires_progress(): + assert _cpu_percent_fields(_cpu_sample(idle=10.0), _cpu_sample(idle=10.0)) is None + + +def test_cpu_percent_fields_skips_on_counter_reset(): + previous = _cpu_sample(idle=100.0, user=50.0) + current = _cpu_sample(idle=200.0, user=1.0) + + assert _cpu_percent_fields(previous, current) is None + + +# -------------------------------------------------------------------------- +# Disk I/O rates +# -------------------------------------------------------------------------- + + +def _sample(timestamp_ns, **counters): + defaults = { + "read_count": 0, + "write_count": 0, + "read_bytes": 0, + "write_bytes": 0, + "read_time": 0, + "write_time": 0, + "busy_time": 0, + } + defaults.update(counters) + defaults["timestamp_ns"] = timestamp_ns + return defaults + + +def test_disk_performance_computes_rates_over_the_interval(): + previous = _sample( + 0, + read_count=100, + write_count=200, + read_bytes=1000, + write_bytes=2000, + read_time=50, + write_time=100, + busy_time=500, + ) + current = _sample( + 2_000_000_000, + read_count=110, + write_count=220, + read_bytes=3000, + write_bytes=6000, + read_time=70, + write_time=140, + busy_time=900, + ) + + fields = _disk_performance_fields(previous, current) + + assert fields == { + "read_bytes_per_sec": (1000.0, "float"), + "write_bytes_per_sec": (2000.0, "float"), + "read_iops": (5.0, "float"), + "write_iops": (10.0, "float"), + "avg_read_latency_ms": (2.0, "float"), + "avg_write_latency_ms": (2.0, "float"), + "util_percent": (20.0, "float"), + } + + +def test_disk_performance_idle_device_reports_zero_latency(): + fields = _disk_performance_fields(_sample(0), _sample(1_000_000_000)) + + assert fields["avg_read_latency_ms"] == (0, "float") + assert fields["read_iops"] == (0.0, "float") + + +def test_disk_performance_requires_a_positive_interval(): + assert _disk_performance_fields(_sample(5), _sample(5)) is None + + +def test_disk_performance_skips_on_counter_reset(): + previous = _sample(0, read_bytes=5000) + current = _sample(1_000_000_000, read_bytes=10) + + assert _disk_performance_fields(previous, current) is None + + +def test_disk_io_sample_defaults_counters_the_platform_omits(): + sample = _disk_io_sample(FakeDiskIO(read_count=1), 42) + + assert sample["read_count"] == 1 + assert sample["busy_time"] == 0 + assert sample["timestamp_ns"] == 42 + + +# -------------------------------------------------------------------------- +# Collectors +# -------------------------------------------------------------------------- + + +def _age_cpu_sample(sample, seconds=100.0): + """Shift a cached sample backwards so the next run sees a non-zero delta.""" + shift = lambda counters: { # noqa: E731 + name: max(0.0, value - seconds) for name, value in counters.items() + } + return { + "total": shift(sample["total"]), + "per_cpu": [shift(core) for core in sample["per_cpu"]], + } + + +def test_cpu_metrics_first_run_reports_no_percentages(client): + lines = collect_cpu_metrics(client, "web-1", "task") + + measurement, tags, fields = parse_line(lines[0].build()) + assert measurement == "system_cpu" + assert tags == {"host": "web-1", "cpu": "total"} + assert {"load1", "load15", "ctx_switches", "syscalls"} <= set(fields) + assert not {"user", "system", "idle", "iowait"} & set(fields) + assert fields["ctx_switches"].endswith("u") # uint64 + assert "No previous CPU sample" in client.messages("info")[0] + assert client.cache.get(_CPU_TIMES_STATE_KEY)["per_cpu"] + + +def test_cpu_metrics_second_run_reports_percentages(client): + collect_cpu_metrics(client, "web-1", "task") + client.cache.put( + _CPU_TIMES_STATE_KEY, _age_cpu_sample(client.cache.get(_CPU_TIMES_STATE_KEY)) + ) + client.logs.clear() + + lines = collect_cpu_metrics(client, "web-1", "task") + + _, _, fields = parse_line(lines[0].build()) + assert {"user", "system", "idle", "iowait", "guest_nice"} <= set(fields) + _, core_tags, core_fields = parse_line(lines[1].build()) + assert measurements(lines[1:]) == ["system_cpu_cores"] * (len(lines) - 1) + assert core_tags == {"host": "web-1", "core": "0"} + assert {"usage", "idle"} <= set(core_fields) + assert not client.logs + + +def test_cpu_metrics_warn_when_per_core_frequency_is_unavailable(client, monkeypatch): + real_cpu_freq = psutil.cpu_freq + + def cpu_freq(percpu=False): + if percpu: + raise NotImplementedError("no per-core frequency") + return real_cpu_freq() + + monkeypatch.setattr(psutil, "cpu_freq", cpu_freq) + + lines = collect_cpu_metrics(client, "web-1", "task") + + # first run: neither percentages nor frequencies, so no core line has fields + assert measurements(lines) == ["system_cpu"] + assert "Error reading per-core CPU frequency" in client.messages("warn")[0] + + +def test_memory_metrics_emit_memory_and_swap(client): + lines = collect_memory_metrics(client, "web-1", "task") + + assert measurements(lines)[:2] == ["system_memory", "system_swap"] + _, tags, fields = parse_line(lines[0].build()) + assert tags == {"host": "web-1"} + assert fields["total"].endswith("u") + assert not fields["percent"].endswith("u") + + +def test_memory_metrics_skip_faults_when_psutil_denies_access(client, monkeypatch): + monkeypatch.setattr(psutil, "Process", raiser(psutil.AccessDenied())) + + lines = collect_memory_metrics(client, "web-1", "task") + + assert measurements(lines) == ["system_memory", "system_swap"] + assert not client.logs + + +def test_network_metrics_emit_one_line_per_interface(client, monkeypatch): + monkeypatch.setattr( + psutil, + "net_io_counters", + lambda pernic=False: { + "eth0": FakeNetIO(1, 2, 3, 4, 5, 6, 7, 8), + "lo": FakeNetIO(9, 10, 11, 12, 13, 14, 15, 16), + }, + ) + + lines = collect_network_metrics(client, "web-1", "task") + + assert measurements(lines) == ["system_network"] * 2 + _, tags, fields = parse_line(lines[0].build()) + assert tags == {"host": "web-1", "interface": "eth0"} + assert fields["bytes_sent"] == "1u" + + +def test_disk_metrics_skips_unreadable_mountpoint(client, monkeypatch): + monkeypatch.setattr( + psutil, + "disk_partitions", + lambda all=False: [ + FakePartition("/dev/sda1", "/", "ext4"), + FakePartition("/dev/sr0", "/media/cdrom", "iso9660"), + ], + ) + + def disk_usage(mountpoint): + if mountpoint == "/": + return FakeUsage(100, 60, 40, 60.0) + raise OSError("No such device") + + monkeypatch.setattr(psutil, "disk_usage", disk_usage) + monkeypatch.setattr(psutil, "disk_io_counters", lambda perdisk=False: {}) + + lines = collect_disk_metrics(client, "web-1", "task") + + assert measurements(lines) == ["system_disk_usage"] + _, tags, _ = parse_line(lines[0].build()) + assert tags["mountpoint"] == "/" + + +def _patch_disk_io(monkeypatch, **counters): + monkeypatch.setattr(psutil, "disk_partitions", lambda all=False: []) + monkeypatch.setattr( + psutil, "disk_io_counters", lambda perdisk=False: {"sda": FakeDiskIO(**counters)} + ) + + +def test_disk_metrics_first_run_caches_counters_without_rates(client, monkeypatch): + _patch_disk_io( + monkeypatch, + read_count=1, + write_count=2, + read_bytes=10, + write_bytes=20, + read_time=3, + write_time=4, + busy_time=5, + read_merged_count=0, + write_merged_count=0, + ) + monkeypatch.setattr(system_metrics.time, "time_ns", lambda: 1_000_000_000) + + lines = collect_disk_metrics(client, "web-1", "task") + + assert measurements(lines) == ["system_disk_io"] + assert "No previous disk I/O sample" in client.messages("info")[0] + cached = client.cache.get(_DISK_IO_STATE_KEY) + assert cached["sda"]["read_bytes"] == 10 + assert client.cache.ttls[_DISK_IO_STATE_KEY] is None + + +def test_disk_metrics_second_run_emits_rates(client, monkeypatch): + client.cache.put( + _DISK_IO_STATE_KEY, + {"sda": _sample(1_000_000_000, read_bytes=10, write_bytes=20)}, + ) + _patch_disk_io( + monkeypatch, + read_count=0, + write_count=0, + read_bytes=1010, + write_bytes=20, + read_time=0, + write_time=0, + busy_time=0, + read_merged_count=0, + write_merged_count=0, + ) + monkeypatch.setattr(system_metrics.time, "time_ns", lambda: 3_000_000_000) + + lines = collect_disk_metrics(client, "web-1", "task") + + assert measurements(lines) == ["system_disk_io", "system_disk_performance"] + _, tags, fields = parse_line(lines[1].build()) + assert tags == {"host": "web-1", "device": "sda"} + assert fields["read_bytes_per_sec"] == "500.0" + assert fields["write_bytes_per_sec"] == "0.0" + assert not client.messages("info") + + +def test_disk_metrics_warns_when_io_counters_fail(client, monkeypatch): + monkeypatch.setattr( + psutil, + "disk_partitions", + lambda all=False: [FakePartition("/dev/sda1", "/", "ext4")], + ) + monkeypatch.setattr(psutil, "disk_usage", lambda mp: FakeUsage(100, 60, 40, 60.0)) + monkeypatch.setattr(psutil, "disk_io_counters", raiser(psutil.Error())) + + lines = collect_disk_metrics(client, "web-1", "task") + + assert measurements(lines) == ["system_disk_usage"] + assert "Error collecting disk I/O metrics" in client.messages("warn")[0] + + +# -------------------------------------------------------------------------- +# Retry wrapper +# -------------------------------------------------------------------------- + + +def test_collect_with_retry_returns_lines_on_first_attempt(client): + def collect(influxdb3_local, hostname, task_id): + return ["line"] + + assert _collect_with_retry(client, collect, "CPU", "web-1", 3, "task") == ["line"] + assert not client.logs + + +def test_collect_with_retry_recovers_after_failures(client): + attempts = [] + + def collect(influxdb3_local, hostname, task_id): + attempts.append(hostname) + if len(attempts) < 3: + raise RuntimeError("flaky") + return ["line"] + + assert _collect_with_retry(client, collect, "disk", "web-1", 3, "task") == ["line"] + assert len(attempts) == 3 + assert len(client.messages("warn")) == 2 + assert "disk metrics collection attempt 1 failed" in client.messages("warn")[0] + + +def test_collect_with_retry_gives_up_after_max_retries(client): + collect = raiser(RuntimeError("always down")) + + assert _collect_with_retry(client, collect, "network", "web-1", 0, "task") is None + assert not client.messages("warn") + assert ( + "Failed to collect network metrics after 0 retries: always down" + in client.messages("error")[0] + ) + + +# -------------------------------------------------------------------------- +# Entry point +# -------------------------------------------------------------------------- + + +def _fake_collectors(calls): + """Mirror the real collector table, recording calls instead of reading psutil.""" + + def make(name): + def collect(influxdb3_local, hostname, task_id): + calls.append((name, hostname)) + return [ + LineBuilder(f"fake_{name}").tag("host", hostname).float64_field("v", 1.0) + ] + + return collect + + collectors = [] + for config_key, metric_type, _ in _COLLECTORS: + name = config_key.removeprefix("include_") + collectors.append((config_key, metric_type, make(name))) + return tuple(collectors) + + +def test_process_runs_every_enabled_collector(client, monkeypatch): + calls = [] + monkeypatch.setattr(system_metrics, "_COLLECTORS", _fake_collectors(calls)) + + process_scheduled_call(client, None, {"hostname": "web-1"}) + + assert [name for name, _ in calls] == ["cpu", "memory", "disk", "network"] + assert all(hostname == "web-1" for _, hostname in calls) + assert client.writes == [ + "fake_cpu,host=web-1 v=1.0", + "fake_memory,host=web-1 v=1.0", + "fake_disk,host=web-1 v=1.0", + "fake_network,host=web-1 v=1.0", + ] + assert "Successfully collected system metrics" in client.messages("info")[-1] + + +def test_process_skips_disabled_collectors(client, monkeypatch): + calls = [] + monkeypatch.setattr(system_metrics, "_COLLECTORS", _fake_collectors(calls)) + + process_scheduled_call( + client, None, {"include_cpu": "false", "include_network": "false"} + ) + + assert [name for name, _ in calls] == ["memory", "disk"] + + +def test_process_writes_nothing_when_config_is_invalid(client, monkeypatch): + calls = [] + monkeypatch.setattr(system_metrics, "_COLLECTORS", _fake_collectors(calls)) + + process_scheduled_call(client, None, {"include_cpu": "maybe"}) + + assert not calls + assert not client.writes + assert client.messages("error") + assert not client.messages("info") + + +def test_process_keeps_other_collectors_when_one_fails(client, monkeypatch): + def cpu(influxdb3_local, hostname, task_id): + return [LineBuilder("fake_cpu").float64_field("v", 1.0)] + + def network(influxdb3_local, hostname, task_id): + return [LineBuilder("fake_net").float64_field("v", 1.0)] + + monkeypatch.setattr( + system_metrics, + "_COLLECTORS", + ( + ("include_cpu", "CPU", cpu), + ("include_disk", "disk", raiser(RuntimeError("disk gone"))), + ("include_network", "network", network), + ), + ) + + process_scheduled_call(client, None, {"max_retries": "0"}) + + assert client.writes == ["fake_cpu v=1.0", "fake_net v=1.0"] + assert ( + "Failed to collect disk metrics after 0 retries: disk gone" + in client.messages("error")[0] + ) + assert "skipped after repeated failures: disk" in client.messages("error")[1] + assert not any("Successfully" in message for message in client.messages("info")) \ No newline at end of file