From 97ab8039fefd8dac0664c7b3a8a1379754ac6275 Mon Sep 17 00:00:00 2001 From: pailozka Date: Wed, 7 Oct 2026 14:39:02 +0200 Subject: [PATCH 1/5] feat: implement plugin execution and pipeline management - Added exec_plugin.py to handle running external plugins as separate processes, ensuring safe execution and JSON input/output. - Introduced kernel.py to manage the plugin pipeline, defining stages (read, score, act, record) and handling plugin resolution and execution. - Refactored controller.py to utilize the new pipeline structure, simplifying evaluation logic and delegating plugin interactions to the pipeline. - Created comprehensive tests in test_pipeline.py to validate the pipeline functionality, including plugin execution, error handling, and configuration management. --- MANIFEST.in | 1 + docs/README.md | 1 + docs/monitoring.md | 10 +- docs/plugins.md | 175 +++++++++++++++ examples/plugins/echo_action.py | 23 ++ examples/plugins/threshold_score.py | 36 +++ src/hpc_eff/config.ini.example | 31 +++ src/hpc_eff/main.py | 20 +- src/hpc_eff/pipeline/__init__.py | 5 + src/hpc_eff/pipeline/builtins.py | 316 ++++++++++++++++++++++++++ src/hpc_eff/pipeline/exec_plugin.py | 75 +++++++ src/hpc_eff/pipeline/kernel.py | 202 +++++++++++++++++ src/hpc_eff/utils/controller.py | 331 +--------------------------- tests/test_pipeline.py | 265 ++++++++++++++++++++++ 14 files changed, 1161 insertions(+), 330 deletions(-) create mode 100644 docs/plugins.md create mode 100755 examples/plugins/echo_action.py create mode 100755 examples/plugins/threshold_score.py create mode 100644 src/hpc_eff/pipeline/__init__.py create mode 100644 src/hpc_eff/pipeline/builtins.py create mode 100644 src/hpc_eff/pipeline/exec_plugin.py create mode 100644 src/hpc_eff/pipeline/kernel.py create mode 100644 tests/test_pipeline.py diff --git a/MANIFEST.in b/MANIFEST.in index d091bb4..ab93ae2 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,2 +1,3 @@ include src/hpc_eff/config.ini.example recursive-include src/hpc_eff/utils *.py *.sql +recursive-include src/hpc_eff/pipeline *.py diff --git a/docs/README.md b/docs/README.md index 5a0900e..86c1768 100644 --- a/docs/README.md +++ b/docs/README.md @@ -7,6 +7,7 @@ | [cluster-rollout.md](cluster-rollout.md) | Many nodes, upgrades, uninstall, first-week health checks. | | [configuration.md](configuration.md) | Every section and key of `/etc/hpc_eff/config.ini`, with defaults and units. | | [regulation-modes.md](regulation-modes.md) | What each mode does with its input: the rating maths, the thermal bands. | +| [plugins.md](plugins.md) | The score/action pipeline: built-in plugins, writing your own, the JSON contract. | | [monitoring.md](monitoring.md) | Database schema, `state.json`, and the SQL worth running once it is live. | | [troubleshooting.md](troubleshooting.md) | Symptom-first: what to check when a step above does not work. | diff --git a/docs/monitoring.md b/docs/monitoring.md index f62cf32..e41f090 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -86,13 +86,14 @@ A JSON file, rewritten in full on every run: "score_name": "SPEC2017", "score_value": 3.8, "power_cmd": "ipmitool dcmi power reading", - "plugins": { "power": "power_reader.py", "price": "energy_price.py", - "co2": "co2_value.py", "temperature": "cpu_thermo.py" } + "plugins": { "read": ["power", "cpufreq"], "score": ["price", "co2", "combine"], + "act": ["cpu_freq"], "record": ["db_log", "state_json"] } }, "current": { "timestamp": "2026-08-21T10:00:03Z", "rating": 5, "rating_price": 6, "rating_co2": 4, "selected_freq_khz": 2600000, + "summary": "Price 2431 is average; Combined rating 5 (price 6, CO2 4)", "action_price": "Price 2431 is average; Combined rating 5 (price 6, CO2 4)", "price": 2431, "co2_current": 412, "co2_median": 448, "co2_grade": 4, "power_w": 287, "cpu_freq_current": 2712 @@ -108,11 +109,14 @@ The example above is a `co2`-mode node. A `temperature`-mode node carries `temperature`, `thermo_freq_limit`, and `action_temp` instead, plus the `gpu_*` fields on GPU nodes; the price and rating keys are absent entirely. -`action_price` (and `action_temp` in `temperature` mode) is the human-readable +`summary` (and `action_temp` in `temperature` mode) is the human-readable summary of what the run decided: the fastest thing to eyeball when something looks wrong. **`action_temp` exists only here, not in the database**, so it is the only place a thermal failure explains itself. +`action_price` is a **deprecated** copy of `summary`, written only when the +`price` plugin runs. It will be removed in a later release; read `summary`. + Each run reads the whole file, replaces `current`, prepends the new entry to `history` (trimmed to `history_length`), refreshes `static`, and writes the whole file back. If the file does not exist yet, it is created with empty diff --git a/docs/plugins.md b/docs/plugins.md new file mode 100644 index 0000000..be14e5c --- /dev/null +++ b/docs/plugins.md @@ -0,0 +1,175 @@ +# Plugins: the score / action pipeline + +Every evaluation is a pipeline of four stages. Each step in a stage is a +plugin, so what gets measured, how it becomes a score, and what is done about +it are separate choices in `config.ini`. + +``` +read ──▶ score ──▶ act ──▶ record + data rating caps DB + state.json + 1–10 +``` + +| Stage | Job | Built-ins | +|---|---|---| +| `read` | Gather raw measurements | `power`, `cpufreq` | +| `score` | Turn data into a `rating` from 1 (run fast) to 10 (throttle most) | `price`, `co2`, `combine` | +| `act` | Do something with the rating | `cpu_freq`, `cpu_thermo`, `gpu_power` | +| `record` | Persist the evaluation | `db_log`, `state_json` | + +Within a stage, plugins run in the order listed. + +## Choosing the pipeline + +**No `[PIPELINE]` section**: the plan follows `[MODE] control_mode` and behaves +exactly as before (`co2` = price + CO₂ rated and combined, then `cpu_freq`; +`temperature` = `cpu_thermo` + `gpu_power`). + +**With `[PIPELINE]`**: it replaces the mode presets. `control_mode` may then be +omitted. A stage you leave out is empty, except `record`, which defaults to +`db_log, state_json`. + +```ini +# Wattnet (or Nowtricity) CO2 alone drives the CPU cap. No price fetch. +[PIPELINE] +score = co2, combine +act = cpu_freq +``` + +`combine` with no `price` plugin simply passes the CO₂ rating through. To pick +the data source, set `[CO2_API] TYPE = wattnet`. + +An invalid pipeline (unknown key, unknown plugin, a plugin listed under the +wrong stage, a bad executable) aborts the run before anything starts, with +`Config error: ...`. No plugin runs and no setting is changed. + +## Your own plugin + +A plugin is any executable. Declare it in a `[plugin:NAME]` section and list +`NAME` in the stage where it belongs: + +```ini +[PIPELINE] +score = co2, threshold_score +act = cpu_freq, echo_action + +[plugin:threshold_score] +exec = /etc/hpc_eff/plugins/threshold_score.py +timeout = 30 # seconds, default 30 +low = 150 # any other key is passed to the plugin as-is +high = 450 + +[plugin:echo_action] +exec = /etc/hpc_eff/plugins/echo_action.py +``` + +A `[plugin:NAME]` with `exec` also **replaces a built-in of the same name**, so +`[plugin:co2] exec = ...` swaps the data source without touching the rest of +the pipeline. `exec = builtin:NAME` selects a built-in explicitly. + +Working examples are in [`examples/plugins/`](../examples/plugins): +`threshold_score.py` (score from absolute gCO₂/kWh thresholds) and +`echo_action.py` (an action that changes nothing: a template to copy). + +### The contract (version 1) + +The controller starts the executable, writes **one JSON request** to its stdin +and reads **one JSON response** from its stdout. + +Request: + +```json +{ + "version": 1, + "stage": "score", + "plugin": "threshold_score", + "config": { "low": "150", "high": "450" }, + "context": { "co2_current": 300, "rating": 5, "...": "everything produced so far" }, + "state": { "timestamp": "2026-10-07T12:00:00Z" } +} +``` + +Response (every field optional): + +```json +{ + "version": 1, + "set": { "rating": 6 }, + "state": { "summary": "CO2 300 g/kWh -> rating 6" } +} +``` + +| Field | Meaning | +|---|---| +| `set` | Merged into the context. Later plugins see it, and `db_log` stores the known columns (`rating`, `co2_current`, `freq_max`, ...). | +| `state` | Merged into this run's `state.json` entry. Any key is allowed. Strings under `summary` from several plugins are joined with `; `. | +| `error` | Non-empty string: report a failure. The run continues. | +| `version` | Contract version the plugin speaks. Omit it or send `1`. A higher number than the controller knows is rejected. | + +`rating`, when set, must be a whole number from 1 to 10. Anything else +rejects that plugin's whole response. + +### Failure handling + +A plugin never aborts the evaluation. Each of these is logged, listed under +`plugin_errors` in the `state.json` entry, and the run goes on with the values +it already has: + +- a non-zero exit code (the last 200 characters of stderr are kept), +- no response within `timeout`, +- output that is not a JSON object, +- an invalid `rating` or unsupported `version`. + +If no plugin scores, `rating` stays at the neutral default of 5 (the same +fallback the built-ins use). + +### Security + +The controller runs as root from cron, so executables are checked **when the +pipeline is built**, and a failing check aborts the run: + +- the path is absolute and the file is executable, +- the file **and its directory** are owned by root (or the user running + `hpc-eff`) and are not writable by group or others, +- it is started without a shell, with working directory `/` and a minimal + environment (`PATH`, `LANG` only; your shell environment is not inherited). + +Put plugins in a root-owned directory such as `/etc/hpc_eff/plugins/` +(`chmod 755`). Configuration values, including anything secret in a +`[plugin:NAME]` section, reach the plugin on stdin, never on the command +line. Keep `config.ini` readable by root only if it holds credentials. + +### Two actions, one knob + +Two `act` plugins can set the same knob (for example `cpu_freq` and +`cpu_thermo` both set the CPU max frequency). The later one wins, and a debug +line names both whenever two actions report a `freq_max`. Pick one per resource. + +## Built-in plugins + +| Name | Stage | Reads | Produces | +|---|---|---|---| +| `power` | read | `[SYSTEM] POWERREADINGCMD` | `power_w` | +| `cpufreq` | read | the running CPU frequency | `cpu_freq_current` | +| `price` | score | electricity price vs. a year of monthly averages | `price`, `rating_price` | +| `co2` | score | `[CO2_API]` (Nowtricity or Wattnet), graded against the last 24 h | `co2_current`, `co2_median`, `co2_grade`, `rating_co2` | +| `combine` | score | `rating_price`, `rating_co2`, `[aggregation]` | `rating` | +| `cpu_freq` | act | `rating`, `[frequency_tables]` | `selected_freq_khz`, `freq_max` | +| `cpu_thermo` | act | `[TEMPERATURE_SOURCE]`, `[CPU_THERMO]` | `temperature`, `freq_max` | +| `gpu_power` | act | `[TEMPERATURE_SOURCE]`, `[GPU_POWER]` | `gpu_*` fields | +| `db_log` | record | the whole context | one row in `hpc_eff_log` | +| `state_json` | record | this run's entry | `state.json` | + +`[aggregation] rating_type` accepts `price`, `co2`, `average` and `max`. +`co2` uses the CO₂ rating alone and falls back to the price rating, then to 5, +when CO₂ data is missing. + +## Limits to know about + +- The database has a fixed set of columns. A custom plugin's extra `set` + values are not stored in SQLite; put anything you want kept under `state` + (it lands in `state.json`). +- The `gpu_power` built-in follows ambient temperature, not the rating. A + GPU cap driven by the rating needs its own action plugin. +- The contract is version 1. New fields will only be added, never changed, so + a plugin written against version 1 keeps working. diff --git a/examples/plugins/echo_action.py b/examples/plugins/echo_action.py new file mode 100755 index 0000000..b868eed --- /dev/null +++ b/examples/plugins/echo_action.py @@ -0,0 +1,23 @@ +#!/usr/bin/env python3 +"""Example action plugin: reports what it would do for the rating. Changes nothing. + +Copy it as a starting point for a real action (lower frequencies, reconfigure +cpufreqd, call a site tool ...). It receives the rating and everything else +gathered so far; whatever it returns under "state" lands in state.json. + +Pipeline use: + [PIPELINE] + act = cpu_freq, echo_action + + [plugin:echo_action] + exec = /etc/hpc_eff/plugins/echo_action.py +""" +import json +import sys + +request = json.load(sys.stdin) +rating = request["context"].get("rating") +print(json.dumps({ + "version": 1, + "state": {"action_echo": f"would act on rating {rating}"}, +})) diff --git a/examples/plugins/threshold_score.py b/examples/plugins/threshold_score.py new file mode 100755 index 0000000..1187799 --- /dev/null +++ b/examples/plugins/threshold_score.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +"""Example score plugin: rating from absolute gCO2/kWh thresholds. + +Reads `co2_current` (set earlier by the built-in `co2` plugin) and maps it +linearly onto 1..10: <= `low` is 1 (run fast), >= `high` is 10 (throttle most). + +Pipeline use: + [PIPELINE] + score = co2, threshold_score + + [plugin:threshold_score] + exec = /etc/hpc_eff/plugins/threshold_score.py + low = 150 + high = 450 +""" +import json +import sys + +request = json.load(sys.stdin) +options = request["config"] +low = float(options.get("low", 100)) +high = float(options.get("high", 500)) +current = request["context"].get("co2_current") + +if current is None: + # No data: say nothing, so the rating set by earlier plugins (or the neutral 5) stands. + print(json.dumps({"version": 1, "set": {}, "state": {"summary": "threshold_score: no CO2 value"}})) + sys.exit(0) + +fraction = min(1.0, max(0.0, (float(current) - low) / (high - low))) +rating = 1 + int(fraction * 9 + 0.5) # half-up, not banker's rounding +print(json.dumps({ + "version": 1, + "set": {"rating": rating}, + "state": {"rating": rating, "summary": f"CO2 {current} g/kWh -> rating {rating} (thresholds {low:g}-{high:g})"}, +})) diff --git a/src/hpc_eff/config.ini.example b/src/hpc_eff/config.ini.example index 2e59714..a5fbcaf 100644 --- a/src/hpc_eff/config.ini.example +++ b/src/hpc_eff/config.ini.example @@ -142,3 +142,34 @@ rating_type=average # higher would let that day-relative volatility dominate the frequency cap. weight_price=0.6 weight_co2=0.4 + +# ============================================================================ +# PLUGIN PIPELINE (optional; replaces the [MODE] control_mode presets) +# ============================================================================ +# Each evaluation runs four stages: read -> score -> act -> record. Every step +# is a plugin: a built-in, or your own executable (see docs/plugins.md). +# Without a [PIPELINE] section the plan follows [MODE] control_mode and behaves +# exactly as before. +# +# Built-ins: read: power, cpufreq | score: price, co2, combine | +# act: cpu_freq, cpu_thermo, gpu_power | record: db_log, state_json +# +# Example: regulate by CO2 alone (Nowtricity or Wattnet per [CO2_API]), no price: +# +#[PIPELINE] +#score = co2, combine +#act = cpu_freq +# +# Example: swap in your own executable. A [plugin:NAME] section with `exec` +# (absolute path, root-owned, not group/world-writable) overrides a built-in +# of the same name or defines a new plugin. Other keys are passed to the plugin. +# +#[PIPELINE] +#score = co2, threshold_score +#act = cpu_freq +# +#[plugin:threshold_score] +#exec = /etc/hpc_eff/plugins/threshold_score.py +#timeout = 30 +#low = 150 +#high = 450 diff --git a/src/hpc_eff/main.py b/src/hpc_eff/main.py index 14b1304..c0a404d 100644 --- a/src/hpc_eff/main.py +++ b/src/hpc_eff/main.py @@ -7,6 +7,7 @@ from .utils.create_log_db import create_log_db from .utils.controller import run_evaluation +from .pipeline import PipelineError from .utils.cron_control import enable_cron, disable_cron # [MODE] control_mode is the SINGLE user-facing switch. It expands into the @@ -45,12 +46,8 @@ "score_name": config.get("SYSTEM", "SCORENAME", fallback="unknown"), "score_value": config.getfloat("SYSTEM", "SCORE", fallback=None), "power_cmd": power_cmd, - "plugins": { - "power": "power_reader.py", - "price": "energy_price.py", - "co2": "co2_value.py", - "temperature": "cpu_thermo.py" - } + # "plugins" (the active plugin names per stage) is filled in by the pipeline + "plugins": {}, } debug = config.get("SYSTEM", "DEBUG", fallback="no").lower() == "yes" @@ -71,6 +68,12 @@ def resolve_control_mode(config): mode = config.get("MODE", "control_mode", fallback=None) if mode: mode = mode.strip().lower() + elif config.has_section("PIPELINE"): + # An explicit [PIPELINE] replaces the control_mode presets entirely. + debug_log("[PIPELINE] present and no control_mode: using the pipeline as configured") + return + if config.has_section("PIPELINE"): + debug_log("[PIPELINE] takes precedence over control_mode; control_mode is ignored") preset = CONTROL_MODE_PRESETS.get(mode) if preset is None: valid = "|".join(CONTROL_MODE_PRESETS) @@ -116,7 +119,10 @@ def main(): # Expand [MODE] control_mode into internal flags, then delegate resolve_control_mode(config) - run_evaluation(conn, config, static_context, debug_log) + try: + run_evaluation(conn, config, static_context, debug_log) + except PipelineError as e: + sys.exit(f"Config error: {e}") if __name__ == "__main__": main() diff --git a/src/hpc_eff/pipeline/__init__.py b/src/hpc_eff/pipeline/__init__.py new file mode 100644 index 0000000..5c7632c --- /dev/null +++ b/src/hpc_eff/pipeline/__init__.py @@ -0,0 +1,5 @@ +"""Plugin pipeline: read -> score -> act -> record. See docs/plugins.md.""" + +from .kernel import PipelineError, build_plan, run_pipeline + +__all__ = ["PipelineError", "build_plan", "run_pipeline"] diff --git a/src/hpc_eff/pipeline/builtins.py b/src/hpc_eff/pipeline/builtins.py new file mode 100644 index 0000000..fe43d0c --- /dev/null +++ b/src/hpc_eff/pipeline/builtins.py @@ -0,0 +1,316 @@ +"""Built-in plugins: today's hardwired behaviour, behind the plugin contract. + +Every built-in is `fn(request, config, runtime) -> {"set": {...}, "state": {...}}`. +`request["context"]` holds the values produced so far. `set` is merged into the +context (and from there into the DB row); `state` is merged into the +state.json entry. Built-ins run in-process and receive the live ConfigParser; +executable plugins get the same request/response as JSON instead. +""" + +import json +import traceback +from pathlib import Path + +from ..utils.energy_price import get_current_energy_price, get_averages_year, classify_price_by_median +from ..utils.power_reader import get_power_reading +from ..utils.frequency_reader import get_cpu_frequency +from ..utils.co2_value import co2_value +from ..utils.cpu_thermo import apply_cpu_thermo +from ..utils.set_cpu import set_cpu_freq, log_setting + +DEFAULT_STATE_PATH = "/var/lib/hpc_eff/state.json" + + +def combine_ratings(rating_price, rating_co2, config, debug_log): + """Combine the price and CO2 ratings into the final rating (1-10). + + Controlled by the [aggregation] section: + - rating_type=price : price rating only (default, legacy behaviour) + - rating_type=co2 : CO2 rating only + - rating_type=average : weighted average of price and CO2 ratings + - rating_type=max : the worse (higher) of the two ratings + + Falls back to whichever rating is available when the preferred one could + not be computed (e.g. CO2 API failure, or a pipeline without a price + plugin), and to the neutral 5 when neither is. + """ + rating_type = config.get("aggregation", "rating_type", fallback="price") + + if rating_type == "co2" and rating_co2 is not None: + return rating_co2 + + if rating_type == "price" or rating_co2 is None: + if rating_type != "price" and rating_co2 is None: + debug_log("Aggregation: CO2 rating unavailable, falling back to price rating") + return rating_price if rating_price is not None else 5 + + if rating_price is None: + debug_log("Aggregation: price rating unavailable, falling back to CO2 rating") + return rating_co2 + + if rating_type == "max": + return max(rating_price, rating_co2) + + if rating_type == "average": + w_price = config.getfloat("aggregation", "weight_price", fallback=0.6) + w_co2 = config.getfloat("aggregation", "weight_co2", fallback=0.4) + combined = (w_price * rating_price + w_co2 * rating_co2) / (w_price + w_co2) + return max(1, min(10, round(combined))) + + debug_log(f"Aggregation: unknown rating_type '{rating_type}', using price rating") + return rating_price + + +def update_state_json(new_entry, state_file_path, history_length, static_context, debug_log): + """Update the JSON state file with history.""" + state_path = Path(state_file_path) + try: + state_path.parent.mkdir(parents=True, exist_ok=True) + + data = {"static": static_context, "history": [], "current": {}} + if state_path.exists(): + try: + with open(state_path, "r") as f: + data = json.load(f) + except (json.JSONDecodeError, IOError): + pass + + data["current"] = new_entry + data["history"].insert(0, new_entry) + data["history"] = data["history"][:history_length] + data["static"] = static_context + + with open(state_path, "w") as f: + json.dump(data, f, indent=4) + + state_path.chmod(0o644) + debug_log(f"Updated state JSON at {state_file_path}") + except Exception as e: + debug_log(f"Error updating state JSON: {e}") + + +# ---------------------------------------------------------------- read stage + +def power(req, config, rt): + """Read the node's instantaneous power draw via [SYSTEM] POWERREADINGCMD.""" + watts = None + try: + data = get_power_reading(config.get("SYSTEM", "POWERREADINGCMD")) + watts = data.get("instantaneous") + rt.debug_log(f"Current power usage (Watts): {watts}") + except Exception as e: + rt.debug_log(f"Error reading power data: {e}") + return {"set": {"power_w": watts}, "state": {"power_w": watts}} + + +def cpufreq(req, config, rt): + """Read the current CPU frequency (MHz).""" + freq = None + try: + freq = get_cpu_frequency() + rt.debug_log(f"Current CPU frequency (MHz): {freq}") + except Exception as e: + rt.debug_log(f"Error reading CPU frequency: {e}") + return {"set": {"cpu_freq_current": freq}, "state": {"cpu_freq_current": freq}} + + +# --------------------------------------------------------------- score stage + +def price(req, config, rt): + """Electricity price, rated against a year of monthly averages.""" + current = None + try: + current = get_current_energy_price() + rt.debug_log(f"Current electricity price (CZK/MWh): {current}") + except Exception as e: + rt.debug_log(f"Error fetching electricity price: {e}") + + averages = [] + try: + averages = get_averages_year() + rt.debug_log(f"Average monthly prices last year (CZK/MWh): {averages}") + except Exception as e: + rt.debug_log(f"Error fetching average prices: {e}") + + rating_price = None + state = {} + try: + classify, rating_price = classify_price_by_median(current, averages) + rt.debug_log(f"The current price of {current} is {classify} ({rating_price}).") + state["summary"] = f"Price {current} is {classify}" + except Exception as e: + rt.debug_log(f"Error fetching classify price and rating: {e}") + + values = {"price": current, "rating_price": rating_price} + state.update(values) + return {"set": values, "state": state} + + +def co2(req, config, rt): + """CO2 intensity (Nowtricity or Wattnet per [CO2_API]), graded 1-10 as a 24 h percentile.""" + try: + history, current, median, grade = co2_value(config) + rt.debug_log(f"Last 24 hours CO2 values (g CO2eq/kWh): {history}") + rt.debug_log(f"Current CO2 value (g CO2eq/kWh): {current}") + rt.debug_log(f"Median value from 24 hours values (g CO2eq/kWh): {median}") + rt.debug_log(f"Current CO2 value grade from 1 (low) to 10 (high): {grade}") + except Exception as e: + current, median, grade = None, None, "unknown" + rt.debug_log(f"Error fetching co2 values and rating: {e}") + + values = { + "co2_current": current, + "co2_median": median, + "co2_grade": grade, + # CO2 grade (1 low - 10 high) doubles as the CO2 rating + "rating_co2": grade if isinstance(grade, int) else None, + } + return {"set": values, "state": dict(values)} + + +def combine(req, config, rt): + """Merge rating_price / rating_co2 into the final `rating` per [aggregation].""" + ctx = req["context"] + rating_price = ctx.get("rating_price") + rating_co2 = ctx.get("rating_co2") + if "rating_price" not in ctx: + # No price plugin ran (e.g. a CO2-only pipeline): there is nothing to + # combine with, so [aggregation] does not apply. + rating = rating_co2 if rating_co2 is not None else 5 + return {"set": {"rating": rating}, "state": {"rating": rating}} + rating = combine_ratings(rating_price, rating_co2, config, rt.debug_log) + + state = {"rating": rating} + if rating != rating_price: + state["summary"] = f"Combined rating {rating} (price {rating_price}, CO2 {rating_co2})" + return {"set": {"rating": rating}, "state": state} + + +# ---------------------------------------------------------------- act stage + +def cpu_freq(req, config, rt): + """Cap the CPU max frequency from `rating` via [frequency_tables].""" + rating = req["context"].get("rating", 5) + rt.debug_log(f"Current rating: {rating}") + selected = set_cpu_freq(rating, config) + + values = {} + state = {"selected_freq_khz": selected} + if selected: + values = {"freq_min": 0, "freq_max": selected} + state["summary"] = f"Rating {rating} set {selected}kHz" + return {"set": values, "state": state} + + +def cpu_thermo(req, config, rt): + """Temperature-banded CPU max frequency with hysteresis ([CPU_THERMO]).""" + notes = [] + temperature = None + values, state = {}, {} + thermo_freq_limit = None + + old_temp = None + try: + state_path = config.get("logging", "state_json_path", fallback=DEFAULT_STATE_PATH) + if Path(state_path).exists(): + with open(state_path, "r") as f: + old_temp = json.load(f).get("current", {}).get("temperature") + except Exception: + pass + + try: + res = apply_cpu_thermo(config) + temperature = res.get("temperature") + values["temperature"] = temperature + if res.get("target_khz") is not None: + values["freq_min"] = 0 + values["freq_max"] = res.get("target_khz") + + status_msg = f"Temp {temperature} C" + if temperature is not None and old_temp is not None: + if temperature >= old_temp + 2: + status_msg += " (rising)" + elif temperature <= old_temp - 2: + status_msg += " (dropping)" + + if res.get("changed"): + thermo_freq_limit = res.get("target_freq") + rt.debug_log(f"cpu_thermo applied target {thermo_freq_limit}") + notes.append(f"{status_msg}: Applied thermal frequency limit: {thermo_freq_limit}") + else: + notes.append(f"{status_msg}: Thermal state stable at {res.get('target_freq')}") + except Exception as e: + rt.debug_log(f"cpu_thermo error: {e}") + notes.append(f"Thermal policy evaluation error: {e}") + + state.update({ + "temperature": temperature, + "thermo_freq_limit": thermo_freq_limit, + "action_temp": "; ".join(notes), + }) + return {"set": values, "state": state} + + +def gpu_power(req, config, rt): + """NVIDIA power limit from ambient temperature ([GPU_POWER]).""" + try: + from ..utils.gpu_power import regulate_gpus + results = regulate_gpus(config, rt.debug_log) + rt.debug_log(f"GPU Power: finished - {len(results)} GPUs processed") + first = results[0] if results else None + except Exception as e: + rt.debug_log(f"GPU Power: error during regulation - {e}") + rt.debug_log(f"GPU Power: traceback: {traceback.format_exc()}") + return {} + + if not first: + return {} + fields = { + "gpu_power_limit": first.get("power_limit"), + "gpu_target_power": first.get("target_power"), + "gpu_state": first.get("state"), + "gpu_changed": first.get("changed", False), + "gpu_success": first.get("success", False), + "gpu_count": first.get("gpu_count"), + } + return {"set": fields, "state": dict(fields)} + + +# ------------------------------------------------------------- record stage + +def db_log(req, config, rt): + """One unified row per evaluation in hpc_eff_log.""" + if rt.conn is None: + return {} + try: + log_setting(rt.conn, **req["context"]) + except Exception as e: + rt.debug_log(f"DB log failed: {e}") + return {} + + +def state_json(req, config, rt): + """Write the evaluation to state.json (current entry plus history).""" + entry = req["state"] + if "summary" in entry and "rating_price" in entry: + # Deprecated alias of `summary`, kept for readers of the old key. + entry["action_price"] = entry["summary"] + path = config.get("logging", "state_json_path", fallback=DEFAULT_STATE_PATH) + history_length = config.getint("logging", "history_length", fallback=10) + update_state_json(entry, path, history_length, rt.static_context, rt.debug_log) + return {} + + +# name -> (stage, function) +BUILTINS = { + "power": ("read", power), + "cpufreq": ("read", cpufreq), + "price": ("score", price), + "co2": ("score", co2), + "combine": ("score", combine), + "cpu_freq": ("act", cpu_freq), + "cpu_thermo": ("act", cpu_thermo), + "gpu_power": ("act", gpu_power), + "db_log": ("record", db_log), + "state_json": ("record", state_json), +} diff --git a/src/hpc_eff/pipeline/exec_plugin.py b/src/hpc_eff/pipeline/exec_plugin.py new file mode 100644 index 0000000..02a6e4b --- /dev/null +++ b/src/hpc_eff/pipeline/exec_plugin.py @@ -0,0 +1,75 @@ +"""Executable transport: run a plugin as a separate process, JSON in / JSON out. + +The plugin receives one JSON request on stdin and must print one JSON +response on stdout (see docs/plugins.md). A non-zero exit, a timeout or +unparsable output is a plugin failure; the kernel isolates it. + +The controller runs as root from cron, so before a plugin is ever started its +file and directory are checked: absolute path, owned by root (or by the user +running the controller), and not writable by group or others. +""" + +import json +import os +import stat +import subprocess + +CONTRACT_VERSION = 1 + +# Plugins never inherit the controller's environment (which may hold secrets). +_SAFE_ENV = {"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C.UTF-8"} + + +class PluginError(Exception): + """A plugin failed at run time (exit code, timeout, bad output).""" + + +def _check_owner_and_mode(path, what): + st = os.stat(path) + if st.st_uid not in (0, os.geteuid()): + raise ValueError(f"{what} {path} must be owned by root (or the user running hpc-eff)") + if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + raise ValueError(f"{what} {path} must not be writable by group or others") + + +def check_executable(path): + """Raise ValueError unless `path` is a safe plugin executable.""" + if not os.path.isabs(path): + raise ValueError(f"plugin path must be absolute: {path!r}") + if not os.path.isfile(path): + raise ValueError(f"plugin not found: {path}") + if not os.access(path, os.X_OK): + raise ValueError(f"plugin is not executable: {path}") + _check_owner_and_mode(path, "plugin") + _check_owner_and_mode(os.path.dirname(path), "plugin directory") + + +def run_executable(path, request, timeout): + """Run the plugin and return its parsed JSON response (a dict).""" + try: + proc = subprocess.run( + [path], + input=json.dumps(request, default=str), + capture_output=True, + text=True, + timeout=timeout, + env=_SAFE_ENV, + cwd="/", + shell=False, + ) + except subprocess.TimeoutExpired: + raise PluginError(f"timed out after {timeout}s") + except OSError as e: + raise PluginError(f"could not start: {e}") + + if proc.returncode != 0: + tail = proc.stderr.strip()[-200:] + raise PluginError(f"exit code {proc.returncode}" + (f": {tail}" if tail else "")) + + try: + response = json.loads(proc.stdout) + except ValueError as e: + raise PluginError(f"stdout is not valid JSON: {e}") + if not isinstance(response, dict): + raise PluginError("stdout JSON must be an object") + return response diff --git a/src/hpc_eff/pipeline/kernel.py b/src/hpc_eff/pipeline/kernel.py new file mode 100644 index 0000000..ab1568e --- /dev/null +++ b/src/hpc_eff/pipeline/kernel.py @@ -0,0 +1,202 @@ +"""Pipeline kernel: builds the plugin plan from config and runs it. + +An evaluation runs four stages in order. Within a stage, plugins run in the +order they are listed: + + read sources of raw measurements (power, cpufreq, ...) + score turn data into a `rating` 1-10 (price, co2, combine, ...) + act do something with the rating (cpu_freq, gpu_power, ...) + record persist the evaluation (db_log, state_json, ...) + +The kernel knows nothing about prices, CO2 or CPUs. Plugins talk to it through +one contract: a request dict in, a response dict out (see docs/plugins.md). +A plugin is either a built-in (in-process) or an executable (JSON over +stdin/stdout). A plugin that fails never aborts the evaluation: the failure is +logged and the run continues with the values it already has. +""" + +import time +from dataclasses import dataclass, field + +from .builtins import BUILTINS +from .exec_plugin import CONTRACT_VERSION, PluginError, check_executable, run_executable + +STAGES = ("read", "score", "act", "record") +DEFAULT_TIMEOUT = 30.0 +DEFAULT_RECORD = ["db_log", "state_json"] +_RESERVED_OPTIONS = ("exec", "timeout") + + +class PipelineError(Exception): + """The pipeline configuration is invalid; the run is aborted before any plugin starts.""" + + +@dataclass +class Plugin: + name: str + exec: str # "builtin:" or an absolute path + timeout: float = DEFAULT_TIMEOUT + options: dict = field(default_factory=dict) + + +@dataclass +class Runtime: + """What built-in plugins may use besides the request and the config.""" + conn: object + debug_log: object + static_context: dict + + +def _names(value): + return [n.strip() for n in value.split(",") if n.strip()] + + +def default_plan(config): + """The plan implied by [FEATURES] (set from [MODE] control_mode), when [PIPELINE] is absent.""" + temp_cpu = config.getboolean("FEATURES", "ENABLE_TEMP_CPU", fallback=True) + price_cpu = config.getboolean("FEATURES", "ENABLE_PRICE_CO2_CPU", fallback=True) + temp_gpu = config.getboolean("FEATURES", "ENABLE_TEMP_GPU", fallback=False) + + return { + "read": ["power", "cpufreq"] if price_cpu else [], + "score": ["price", "co2", "combine"] if price_cpu else [], + "act": (["cpu_thermo"] if temp_cpu else []) + (["cpu_freq"] if price_cpu else []) + + (["gpu_power"] if temp_gpu else []), + "record": list(DEFAULT_RECORD), + } + + +def explicit_plan(config): + """The plan named by [PIPELINE]. `record` defaults to db_log, state_json.""" + section = config["PIPELINE"] + unknown = [k for k in section if k not in STAGES] + if unknown: + raise PipelineError( + f"[PIPELINE] unknown key(s) {unknown}; valid keys: {', '.join(STAGES)}" + ) + plan = {stage: _names(section.get(stage, "")) for stage in STAGES} + if "record" not in section: + plan["record"] = list(DEFAULT_RECORD) + return plan + + +def resolve(name, stage, config): + """Turn a plugin name into a Plugin, validating it. Raises PipelineError.""" + options, target, timeout = {}, None, DEFAULT_TIMEOUT + section = f"plugin:{name}" + if config.has_section(section): + options = dict(config.items(section, raw=True)) + target = options.get("exec") + try: + timeout = float(options.get("timeout", DEFAULT_TIMEOUT)) + except ValueError: + raise PipelineError(f"[{section}] timeout must be a number") + for key in _RESERVED_OPTIONS: + options.pop(key, None) + + if not target: + target = f"builtin:{name}" + + if target.startswith("builtin:"): + builtin = target[len("builtin:"):] + if builtin not in BUILTINS: + raise PipelineError( + f"unknown plugin {name!r}: no built-in {builtin!r} " + f"(built-ins: {', '.join(sorted(BUILTINS))}) and no exec path in [{section}]" + ) + if BUILTINS[builtin][0] != stage: + raise PipelineError( + f"plugin {name!r} is a '{BUILTINS[builtin][0]}' plugin, but is listed under '{stage}'" + ) + else: + try: + check_executable(target) + except ValueError as e: + raise PipelineError(f"plugin {name!r}: {e}") + + return Plugin(name=name, exec=target, timeout=timeout, options=options) + + +def build_plan(config): + """Return {stage: [Plugin, ...]} for this run. Raises PipelineError.""" + names = explicit_plan(config) if config.has_section("PIPELINE") else default_plan(config) + return {stage: [resolve(n, stage, config) for n in names[stage]] for stage in STAGES} + + +def _valid_rating(value): + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise PluginError(f"rating must be a number from 1 to 10, got {value!r}") + if value != int(value) or not 1 <= value <= 10: + raise PluginError(f"rating must be a whole number from 1 to 10, got {value!r}") + return int(value) + + +def _apply(response, ctx, entry): + """Validate a plugin response and merge it into the context and state entry.""" + version = response.get("version", CONTRACT_VERSION) + if not isinstance(version, int) or version > CONTRACT_VERSION: + raise PluginError(f"unsupported contract version {version!r} (this controller speaks {CONTRACT_VERSION})") + values = response.get("set", {}) + state = response.get("state", {}) + if not isinstance(values, dict) or not isinstance(state, dict): + raise PluginError("'set' and 'state' must be JSON objects") + if response.get("error"): + raise PluginError(str(response["error"])) + + values = dict(values) + if "rating" in values: + values["rating"] = _valid_rating(values["rating"]) + + ctx.update(values) + for key, val in state.items(): + # `summary` notes from several plugins are joined, everything else overwrites + if key == "summary" and isinstance(entry.get(key), str) and isinstance(val, str): + entry[key] = f"{entry[key]}; {val}" + else: + entry[key] = val + + +def _call(plugin, request, config, runtime): + if plugin.exec.startswith("builtin:"): + return BUILTINS[plugin.exec[len("builtin:"):]][1](request, config, runtime) or {} + return run_executable(plugin.exec, request, plugin.timeout) + + +def run_pipeline(conn, config, static_context, debug_log): + """Run one evaluation. Raises PipelineError only for an invalid configuration.""" + plan = build_plan(config) + names = {stage: [p.name for p in plan[stage]] for stage in STAGES} + debug_log(f"Pipeline: {names}") + + static = dict(static_context, plugins=names) + runtime = Runtime(conn=conn, debug_log=debug_log, static_context=static) + + ctx: dict = dict(static) + ctx.update({"rating": 5, "temperature": None}) # neutral default until a plugin scores + entry: dict = {"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} + errors = [] + freq_setter = None + + for stage in STAGES: + if stage == "record" and errors: + entry["plugin_errors"] = errors + for plugin in plan[stage]: + request = { + "version": CONTRACT_VERSION, + "stage": stage, + "plugin": plugin.name, + "config": plugin.options, + "context": ctx, + "state": entry, + } + try: + response = _call(plugin, request, config, runtime) + _apply(response, ctx, entry) + if stage == "act" and "freq_max" in response.get("set", {}): + if freq_setter: + debug_log(f"Pipeline: {plugin.name} overrides the CPU max frequency set by {freq_setter}") + freq_setter = plugin.name + except Exception as e: + message = f"{plugin.name}: {e}" + errors.append(message) + debug_log(f"Pipeline: plugin failed - {message}") diff --git a/src/hpc_eff/utils/controller.py b/src/hpc_eff/utils/controller.py index df875f4..970f60d 100644 --- a/src/hpc_eff/utils/controller.py +++ b/src/hpc_eff/utils/controller.py @@ -1,91 +1,18 @@ -"""Controller/orchestrator for evaluation and action calls. +"""Controller entry point: runs one evaluation through the plugin pipeline. -Moves logic out of `main.py`: gathers inputs (price, power, freq, CO2, -temperature), computes ratings and log context, and conditionally calls the -terminal actions `apply_cpu_thermo` and `set_cpu_freq` based on config flags. +The evaluation itself lives in `hpc_eff.pipeline` (read -> score -> act -> +record, each step a plugin; see docs/plugins.md). This module keeps the +long-standing `run_evaluation` entry point. -Internal regulator flags drive the pipeline. These live in an internal -[FEATURES] section that `main.resolve_control_mode` populates from the single -user-facing `[MODE] control_mode` switch (the user never sets them directly): +Without a [PIPELINE] section, the plan is derived from the internal [FEATURES] +flags that `main.resolve_control_mode` populates from `[MODE] control_mode`: - ENABLE_PRICE_CO2_CPU : price/CO2 -> rating -> max-frequency cap - ENABLE_TEMP_CPU : temperature -> hysteresis-based max-frequency control - ENABLE_TEMP_GPU : temperature -> power limiting for NVIDIA GPUs - -control_mode=co2 sets PRICE_CPU; control_mode=temperature sets TEMP_CPU + -TEMP_GPU. CPU regulation is mutually exclusive, so the two never run together. """ -import json -import time -from pathlib import Path - -from .energy_price import get_current_energy_price, get_averages_year, classify_price_by_median -from .power_reader import get_power_reading -from .frequency_reader import get_cpu_frequency -from .co2_value import co2_value -from .cpu_thermo import apply_cpu_thermo -from .set_cpu import set_cpu_freq, log_setting - -def combine_ratings(rating_price, rating_co2, config, debug_log): - """Combine the price and CO2 ratings into the final rating (1-10). - - Controlled by the [aggregation] section: - - rating_type=price : price rating only (default, legacy behaviour) - - rating_type=average : weighted average of price and CO2 ratings - - rating_type=max : the worse (higher) of the two ratings - - Falls back to whichever rating is available when the other could not be - computed (e.g. CO2 API failure), and to the neutral 5 when neither is. - """ - rating_type = config.get("aggregation", "rating_type", fallback="price") - - if rating_type == "price" or rating_co2 is None: - if rating_type != "price" and rating_co2 is None: - debug_log("Aggregation: CO2 rating unavailable, falling back to price rating") - return rating_price if rating_price is not None else 5 - - if rating_price is None: - debug_log("Aggregation: price rating unavailable, falling back to CO2 rating") - return rating_co2 - - if rating_type == "max": - return max(rating_price, rating_co2) - - if rating_type == "average": - w_price = config.getfloat("aggregation", "weight_price", fallback=0.6) - w_co2 = config.getfloat("aggregation", "weight_co2", fallback=0.4) - combined = (w_price * rating_price + w_co2 * rating_co2) / (w_price + w_co2) - return max(1, min(10, round(combined))) +from ..pipeline import run_pipeline - debug_log(f"Aggregation: unknown rating_type '{rating_type}', using price rating") - return rating_price - -def update_state_json(new_entry, state_file_path, history_length, static_context, debug_log): - """Update the JSON state file with history.""" - state_path = Path(state_file_path) - try: - state_path.parent.mkdir(parents=True, exist_ok=True) - - data = {"static": static_context, "history": [], "current": {}} - if state_path.exists(): - try: - with open(state_path, "r") as f: - data = json.load(f) - except (json.JSONDecodeError, IOError): - pass - - data["current"] = new_entry - data["history"].insert(0, new_entry) - data["history"] = data["history"][:history_length] - data["static"] = static_context - - with open(state_path, "w") as f: - json.dump(data, f, indent=4) - - state_path.chmod(0o644) - debug_log(f"Updated state JSON at {state_file_path}") - except Exception as e: - debug_log(f"Error updating state JSON: {e}") def run_evaluation(conn, config, static_context: dict, debug_log): """Run the full evaluation pipeline and apply actions depending on config. @@ -96,246 +23,10 @@ def run_evaluation(conn, config, static_context: dict, debug_log): static_context: dict with static values (score_name, score_value) debug_log: callable for debug logging - Reads the internal flags `FEATURES/ENABLE_TEMP_CPU` and - `FEATURES/ENABLE_PRICE_CO2_CPU` (set from `[MODE] control_mode`) to decide which - final action to call (they are mutually exclusive). Data that a disabled - action would need is not gathered, to save work. + Raises hpc_eff.pipeline.PipelineError if the pipeline configuration is + invalid. Plugin failures at run time never raise: they are logged and the + evaluation continues. """ debug_log("Controller: starting evaluation") - - # Determine which final actions are enabled; avoid gathering data - # that we don't need when the corresponding action is disabled. - enable_temp_cpu = config.getboolean("FEATURES", "ENABLE_TEMP_CPU", fallback=True) - enable_price_cpu = config.getboolean("FEATURES", "ENABLE_PRICE_CO2_CPU", fallback=True) - enable_temp_gpu = config.getboolean("FEATURES", "ENABLE_TEMP_GPU", fallback=False) - - # initialize placeholders - price = power_w = cpu_freq_current = temperature = None - available_freqs = available_govs = average_prices = historical_values = [] - current_price = None - rating = 5 # neutral default if not computed - rating_price = rating_co2 = None - current_value = median_value = grade = None - price_notes = [] - temp_notes = [] - - # GPU power tracking - gpu_power_result = None - - # Build base log_context with static info - log_context = static_context.copy() - - # If the price regulator is enabled we need price/history/CO2/power/freq to compute rating - if enable_price_cpu: - # Read current electricity price (CZK/MWh) - try: - price = get_current_energy_price() - debug_log(f"Current electricity price (CZK/MWh): {price}") - except Exception as e: - debug_log(f"Error fetching electricity price: {e}") - - # Read current power consumption - try: - power_cmd = config.get("SYSTEM", "POWERREADINGCMD") - power_data = get_power_reading(power_cmd) - power_w = power_data.get("instantaneous") - debug_log(f"Current power usage (Watts): {power_w}") - except Exception as e: - debug_log(f"Error reading power data: {e}") - - # Read current CPU frequency - try: - cpu_freq_current = get_cpu_frequency() - debug_log(f"Current CPU frequency (MHz): {cpu_freq_current}") - except Exception as e: - debug_log(f"Error reading CPU frequency: {e}") - - # Get year average prices (used for both logging and classification) - try: - average_prices = get_averages_year() - debug_log(f"Average monthly prices last year (CZK/MWh): {average_prices}") - except Exception as e: - debug_log(f"Error fetching average prices: {e}") - - # Determine current_price - current_price = price - - # Classify and rate the current price against the same year history - try: - classify, rating = classify_price_by_median(current_price, average_prices) - rating_price = rating - debug_log(f"The current price of {current_price} is {classify} ({rating}).") - price_notes.append(f"Price {current_price} is {classify}") - except Exception as e: - debug_log(f"Error fetching classify price and rating: {e}") - - # Fetch CO2 values and grading (only if the price regulator needs them for logging) - try: - # Pass config object to support both nowtricity and wattnet APIs - historical_values, current_value, median_value, grade = co2_value(config) - debug_log(f"Last 24 hours CO2 values (g CO2eq/kWh): {historical_values}") - debug_log(f"Current CO2 value (g CO2eq/kWh): {current_value}") - debug_log(f"Median value from 24 hours values (g CO2eq/kWh): {median_value}") - debug_log(f"Current CO2 value grade from 1 (low) to 10 (high): {grade}") - except Exception as e: - historical_values, current_value, median_value, grade = None, None, None, "unknown" - debug_log(f"Error fetching co2 values and rating: {e}") - - # CO2 grade (1 low - 10 high) doubles as the CO2 rating - rating_co2 = grade if isinstance(grade, int) else None - - # Combine price and CO2 ratings per the [aggregation] config - rating = combine_ratings(rating_price, rating_co2, config, debug_log) - if rating != rating_price: - price_notes.append(f"Combined rating {rating} (price {rating_price}, CO2 {rating_co2})") - - # update log_context with gathered values - log_context.update({ - "price": current_price, - "co2_current": current_value, - "co2_median": median_value, - "co2_grade": grade, - "rating_price": rating_price, - "rating_co2": rating_co2, - "power_w": power_w, - "cpu_freq_current": cpu_freq_current, - }) - - # Ensure core metrics are in log_context for other functions - log_context.update({ - "rating": rating, - "temperature": temperature, - }) - - # Get previous temperature for comparison (rising/dropping annotation) - old_temp = None - try: - state_file_path = config.get("logging", "state_json_path", fallback="/var/lib/hpc_eff/state.json") - if Path(state_file_path).exists(): - with open(state_file_path, "r") as f: - old_data = json.load(f) - old_temp = old_data.get("current", {}).get("temperature") - except Exception: - pass - - # Apply temperature-based CPU frequency control - thermo_freq_limit = None - if enable_temp_cpu: - try: - res = apply_cpu_thermo(config) - temperature = res.get("temperature") - log_context["temperature"] = temperature # update with real reading - # Record the resulting max-frequency cap for the unified DB row - if res.get("target_khz") is not None: - log_context["freq_min"] = 0 - log_context["freq_max"] = res.get("target_khz") - - status_msg = f"Temp {temperature} C" - if temperature is not None and old_temp is not None: - if temperature >= old_temp + 2: - status_msg += " (rising)" - elif temperature <= old_temp - 2: - status_msg += " (dropping)" - - if res.get("changed"): - thermo_freq_limit = res.get('target_freq') - debug_log(f"cpu_thermo applied target {thermo_freq_limit}") - temp_notes.append(f"{status_msg}: Applied thermal frequency limit: {thermo_freq_limit}") - else: - current_target = res.get('target_freq') - temp_notes.append(f"{status_msg}: Thermal state stable at {current_target}") - except Exception as e: - debug_log(f"cpu_thermo error: {e}") - temp_notes.append(f"Thermal policy evaluation error: {e}") - else: - debug_log("CPU thermo disabled; skipping") - - # NOTE: CPU regulation is mutually exclusive (temperature OR price), enforced - # in main.validate_features. Price mode does not read temperature and has no - # thermal cutoff here; temperature mode handles heat via [CPU_THERMO] bands. - - # Set CPU max frequency based on rating - selected_freq = None - if enable_price_cpu: - debug_log(f"Current rating: {rating}") - selected_freq = set_cpu_freq(rating, config, **log_context) - if selected_freq: - # Record the resulting max-frequency cap for the unified DB row - log_context["freq_min"] = 0 - log_context["freq_max"] = selected_freq - price_notes.append(f"Rating {rating} set {selected_freq}kHz") - else: - debug_log("set_cpu_freq disabled; skipping") - - # Finalize log entry for JSON (no static info) - json_entry = { - "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - } - - # Temperature/thermo fields only if the thermo action is enabled - if enable_temp_cpu: - json_entry.update({ - "temperature": temperature, - "thermo_freq_limit": thermo_freq_limit, - "action_temp": "; ".join(temp_notes), - }) - - # Price/rating/frequency fields only if the price regulator is enabled - if enable_price_cpu: - json_entry.update({ - "rating": rating, - "rating_price": rating_price, - "rating_co2": rating_co2, - "selected_freq_khz": selected_freq, - "action_price": "; ".join(price_notes), - "price": current_price, - "co2_current": current_value, - "co2_median": median_value, - "co2_grade": grade, - "power_w": power_w, - "cpu_freq_current": cpu_freq_current, - }) - - # GPU Power Regulation (MUST run BEFORE json_entry.update for GPU fields) - if enable_temp_gpu: - try: - from .gpu_power import regulate_gpus as gpu_regulate - results = gpu_regulate(config, debug_log) - debug_log(f"GPU Power: finished - {len(results)} GPUs processed") - if results and len(results) > 0: - gpu_power_result = results[0] # Take first (and only) result - except ImportError as e: - debug_log(f"GPU Power: module not found - {e}") - except Exception as e: - import traceback - debug_log(f"GPU Power: error during regulation - {e}") - debug_log(f"GPU Power: traceback: {traceback.format_exc()}") - else: - debug_log("GPU Power: disabled (FEATURES/ENABLE_TEMP_GPU=no)") - - # GPU power fields only if GPU power is enabled AND we have a result - if enable_temp_gpu and gpu_power_result: - gpu_fields = { - "gpu_power_limit": gpu_power_result.get("power_limit"), - "gpu_target_power": gpu_power_result.get("target_power"), - "gpu_state": gpu_power_result.get("state"), - "gpu_changed": gpu_power_result.get("changed", False), - "gpu_success": gpu_power_result.get("success", False), - "gpu_count": gpu_power_result.get("gpu_count"), - } - json_entry.update(gpu_fields) - log_context.update(gpu_fields) - - # Single unified DB write: one row per evaluation with all CPU + GPU data. - if conn is not None: - try: - log_setting(conn, **log_context) - except Exception as e: - debug_log(f"DB log failed: {e}") - - # Update state JSON - state_file_path = config.get("logging", "state_json_path", fallback="/var/lib/hpc_eff/state.json") - history_length = config.getint("logging", "history_length", fallback=10) - update_state_json(json_entry, state_file_path, history_length, static_context, debug_log) - + run_pipeline(conn, config, static_context, debug_log) debug_log("Controller: evaluation finished") diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..2a8a477 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,265 @@ +"""Tests for the plugin pipeline and the Wattnet parsing fix. + +Run from the repo root: PYTHONPATH=src python3 -m unittest discover -s tests -v + +Nothing here touches the real system: cpupower/nvidia-smi calls, the network +and the config under /etc are all mocked or redirected into temp directories. +""" +import configparser +import json +import os +import shutil +import sqlite3 +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest import mock + +from hpc_eff.pipeline import PipelineError, build_plan, run_pipeline +from hpc_eff.pipeline import builtins as pb +from hpc_eff.utils.create_log_db import create_log_db +from hpc_eff.utils import co2_value as co2_mod + +EXAMPLES = Path(__file__).resolve().parent.parent / "examples" / "plugins" + + +def make_config(text): + cfg = configparser.ConfigParser() + cfg.read_string(text) + return cfg + + +class TempDirCase(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + os.chmod(self.tmp, 0o755) + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.plugins = self.tmp / "plugins" + self.plugins.mkdir(mode=0o755) + os.chmod(self.plugins, 0o755) + + def plugin(self, name, body, mode=0o755): + path = self.plugins / name + path.write_text("#!/usr/bin/env python3\n" + body) + os.chmod(path, mode) + return str(path) + + def example(self, name): + dest = self.plugins / name + shutil.copy(EXAMPLES / name, dest) + os.chmod(dest, 0o755) + return str(dest) + + +def quiet(_message): + pass + + +class WattnetParsingTest(unittest.TestCase): + """fetch_wattnet_24h must treat the NEWEST hour as 'current'.""" + + def test_current_is_newest_hour(self): + end = datetime.now(timezone.utc).replace(microsecond=0) + samples = [] + for i in range(96): # 96 x 15 min = 24 h, rising 100 -> 195 + ts = (end - timedelta(minutes=15 * (95 - i))).strftime("%Y-%m-%dT%H:%M:%SZ") + samples.append([ts, 100 + i]) + response = mock.Mock() + response.json.return_value = [{"series": [{"values": samples}]}] + response.raise_for_status.return_value = None + cfg = make_config("[CO2_API]\nTYPE=wattnet\nWATTNET_API_KEY=x\n") + + with mock.patch.object(co2_mod.requests, "get", return_value=response): + history, current, _median, grade = co2_mod.co2_value(cfg) + + self.assertEqual(len(history), 24) + self.assertEqual(current, history[0]) + self.assertGreater(current, 190) # newest hour (was ~102 before the fix) + self.assertEqual(grade, 10) + + +class CombineRatingsTest(unittest.TestCase): + def cfg(self, rating_type): + return make_config(f"[aggregation]\nrating_type={rating_type}\n") + + def test_co2_only(self): + self.assertEqual(pb.combine_ratings(2, 9, self.cfg("co2"), quiet), 9) + + def test_co2_falls_back_to_price_then_neutral(self): + self.assertEqual(pb.combine_ratings(4, None, self.cfg("co2"), quiet), 4) + self.assertEqual(pb.combine_ratings(None, None, self.cfg("co2"), quiet), 5) + + def test_existing_types_unchanged(self): + self.assertEqual(pb.combine_ratings(2, 9, self.cfg("max"), quiet), 9) + self.assertEqual(pb.combine_ratings(2, 9, self.cfg("price"), quiet), 2) + self.assertEqual(pb.combine_ratings(5, 10, self.cfg("average"), quiet), 7) # (0.6*5+0.4*10)/1 + + +class PlanTest(TempDirCase): + def names(self, cfg): + plan = build_plan(cfg) + return {stage: [p.name for p in plugins] for stage, plugins in plan.items()} + + def test_default_co2_mode(self): + cfg = make_config("[FEATURES]\nENABLE_TEMP_CPU=no\nENABLE_PRICE_CO2_CPU=yes\nENABLE_TEMP_GPU=no\n") + self.assertEqual(self.names(cfg), { + "read": ["power", "cpufreq"], "score": ["price", "co2", "combine"], + "act": ["cpu_freq"], "record": ["db_log", "state_json"]}) + + def test_default_temperature_mode(self): + cfg = make_config("[FEATURES]\nENABLE_TEMP_CPU=yes\nENABLE_PRICE_CO2_CPU=no\nENABLE_TEMP_GPU=yes\n") + self.assertEqual(self.names(cfg), { + "read": [], "score": [], "act": ["cpu_thermo", "gpu_power"], + "record": ["db_log", "state_json"]}) + + def test_explicit_pipeline_record_defaults(self): + cfg = make_config("[PIPELINE]\nscore = co2, combine\nact = cpu_freq, gpu_power\n") + names = self.names(cfg) + self.assertEqual(names["read"], []) + self.assertEqual(names["record"], ["db_log", "state_json"]) + + def test_executable_overrides_builtin_name(self): + path = self.example("threshold_score.py") + cfg = make_config(f"[PIPELINE]\nscore = co2\n[plugin:co2]\nexec = {path}\n") + self.assertEqual(build_plan(cfg)["score"][0].exec, path) + + def test_config_errors_abort(self): + bad = { + "unknown key": "[PIPELINE]\nacts = cpu_freq\n", + "unknown plugin": "[PIPELINE]\nact = nope\n", + "wrong stage": "[PIPELINE]\nscore = cpu_freq\n", + "relative path": "[PIPELINE]\nscore = x\n[plugin:x]\nexec = plugins/x.py\n", + "missing file": "[PIPELINE]\nscore = x\n[plugin:x]\nexec = /nonexistent/x.py\n", + "bad timeout": "[PIPELINE]\nscore = co2\n[plugin:co2]\ntimeout = soon\n", + } + for label, text in bad.items(): + with self.subTest(label), self.assertRaises(PipelineError): + build_plan(make_config(text)) + + def test_unsafe_permissions_rejected(self): + path = self.plugin("w.py", "pass\n", mode=0o775) # group-writable + with self.assertRaises(PipelineError): + build_plan(make_config(f"[PIPELINE]\nscore = w\n[plugin:w]\nexec = {path}\n")) + path = self.plugin("w2.py", "pass\n", mode=0o755) + os.chmod(self.plugins, 0o777) # world-writable dir + self.addCleanup(os.chmod, self.plugins, 0o755) + with self.assertRaises(PipelineError): + build_plan(make_config(f"[PIPELINE]\nscore = w2\n[plugin:w2]\nexec = {path}\n")) + + def test_not_executable_rejected(self): + path = self.plugin("n.py", "pass\n", mode=0o644) + with self.assertRaises(PipelineError): + build_plan(make_config(f"[PIPELINE]\nscore = n\n[plugin:n]\nexec = {path}\n")) + + +class RunPipelineTest(TempDirCase): + def setUp(self): + super().setUp() + self.db = self.tmp / "history.db" + create_log_db(self.db) + self.conn = sqlite3.connect(self.db) + self.addCleanup(self.conn.close) + self.state = self.tmp / "state.json" + self.logs = [] + + def run_cfg(self, text, **patches): + cfg = make_config(f"[logging]\nstate_json_path={self.state}\nhistory_length=5\n" + text) + static = {"score_name": "t", "score_value": 1.0, "power_cmd": "", "plugins": {}} + co2 = patches.get("co2", (list(range(24)), 300, 150, 7)) + co2_patch = {"side_effect": co2} if isinstance(co2, Exception) else {"return_value": co2} + with mock.patch.object(pb, "co2_value", **co2_patch), \ + mock.patch.object(pb, "set_cpu_freq", return_value=2200000) as set_freq: + run_pipeline(self.conn, cfg, static, self.logs.append) + self.set_freq = set_freq + + def entry(self): + return json.loads(self.state.read_text())["current"] + + def row(self): + self.conn.row_factory = sqlite3.Row + return dict(self.conn.execute("SELECT * FROM hpc_eff_log").fetchone()) + + def test_co2_only_pipeline_fetches_no_price(self): + with mock.patch.object(pb, "get_current_energy_price", side_effect=AssertionError("price fetched")): + self.run_cfg("[PIPELINE]\nscore = co2, combine\nact = cpu_freq\n") + self.set_freq.assert_called_once() + self.assertEqual(self.set_freq.call_args[0][0], 7) # rating = CO2 grade + entry = self.entry() + self.assertEqual(entry["rating"], 7) + self.assertEqual(entry["co2_current"], 300) + self.assertEqual(entry["selected_freq_khz"], 2200000) + self.assertNotIn("price", entry) + self.assertNotIn("plugin_errors", entry) + self.assertEqual(self.row()["rating"], 7) + self.assertEqual(self.row()["freq_max"], 2200000) + static = json.loads(self.state.read_text())["static"] + self.assertEqual(static["plugins"]["score"], ["co2", "combine"]) + + def test_co2_failure_falls_back_to_neutral_rating(self): + self.run_cfg("[PIPELINE]\nscore = co2, combine\nact = cpu_freq\n", co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 5) + self.assertEqual(self.entry()["co2_grade"], "unknown") + + def test_executable_score_plugin_sets_rating(self): + path = self.example("threshold_score.py") + self.run_cfg(f"[PIPELINE]\nscore = co2, threshold_score\nact = cpu_freq\n" + f"[plugin:threshold_score]\nexec = {path}\nlow = 100\nhigh = 500\n") + # 300 g/kWh between 100 and 500 -> halfway -> 1 + round(4.5) = 6 + self.assertEqual(self.set_freq.call_args[0][0], 6) + self.assertIn("rating 6", self.entry()["summary"]) + + def test_executable_action_plugin_receives_rating(self): + path = self.example("echo_action.py") + self.run_cfg(f"[PIPELINE]\nscore = co2, combine\nact = echo_action\n" + f"[plugin:echo_action]\nexec = {path}\n") + self.assertEqual(self.entry()["action_echo"], "would act on rating 7") + self.set_freq.assert_not_called() + + def test_failing_plugins_are_isolated(self): + crash = self.plugin("crash.py", "import sys\nsys.stderr.write('boom')\nsys.exit(3)\n") + slow = self.plugin("slow.py", "import time\ntime.sleep(5)\n") + junk = self.plugin("junk.py", "print('not json')\n") + wild = self.plugin("wild.py", "import json\nprint(json.dumps({'set': {'rating': 11}}))\n") + future = self.plugin("future.py", "import json\nprint(json.dumps({'version': 2}))\n") + names = ["crash", "slow", "junk", "wild", "future"] + paths = dict(zip(names, [crash, slow, junk, wild, future])) + sections = "".join(f"[plugin:{n}]\nexec = {p}\ntimeout = 0.5\n" for n, p in paths.items()) + self.run_cfg(f"[PIPELINE]\nscore = co2, combine, {', '.join(names)}\nact = cpu_freq\n" + sections) + errors = self.entry()["plugin_errors"] + self.assertEqual(len(errors), 5) + self.assertIn("exit code 3: boom", errors[0]) + self.assertIn("timed out", errors[1]) + self.assertIn("not valid JSON", errors[2]) + self.assertIn("rating must be", errors[3]) + self.assertIn("unsupported contract version", errors[4]) + # the evaluation still completed with the rating computed before the failures + self.assertEqual(self.set_freq.call_args[0][0], 7) + self.assertEqual(self.row()["rating"], 7) + + def test_plugin_environment_is_scrubbed(self): + probe = self.plugin( + "env.py", + "import json, os\nprint(json.dumps({'state': {'leak': os.environ.get('HPC_EFF_SECRET', 'none')}}))\n") + with mock.patch.dict(os.environ, {"HPC_EFF_SECRET": "hunter2"}): + self.run_cfg(f"[PIPELINE]\nscore = env\n[plugin:env]\nexec = {probe}\n") + self.assertEqual(self.entry()["leak"], "none") + + def test_default_plan_still_runs_price_and_cpu_freq(self): + with mock.patch.object(pb, "get_current_energy_price", return_value=2000.0), \ + mock.patch.object(pb, "get_averages_year", return_value=[1500.0] * 12), \ + mock.patch.object(pb, "get_power_reading", return_value={"instantaneous": 321}), \ + mock.patch.object(pb, "get_cpu_frequency", return_value=2400): + self.run_cfg("[SYSTEM]\nPOWERREADINGCMD=true\n" + "[FEATURES]\nENABLE_TEMP_CPU=no\nENABLE_PRICE_CO2_CPU=yes\nENABLE_TEMP_GPU=no\n" + "[aggregation]\nrating_type=average\n") + entry = self.entry() + for key in ("price", "rating_price", "rating_co2", "rating", "power_w", "cpu_freq_current", + "selected_freq_khz", "co2_current", "summary", "action_price"): + self.assertIn(key, entry) + self.assertEqual(entry["power_w"], 321) + self.assertEqual(self.row()["power_w"], 321) + + +if __name__ == "__main__": + unittest.main() From 5b78d0bdb6ee858e23779f9be32127612636823c Mon Sep 17 00:00:00 2001 From: pailozka Date: Wed, 7 Oct 2026 14:46:44 +0200 Subject: [PATCH 2/5] feat: plugin starter, packaged examples, docs; fix deb build dependency - add examples/plugins/template.py, a blank starter for new plugins - install examples/plugins under the package doc dir (deb and rpm) and include them in the sdist - document plugins in README.md, docs/configuration.md ([PIPELINE] and [plugin:NAME]) and docs/plugins.md - add python3-requests to Build-Depends: dh_auto_test imports hpc_eff.pipeline, which imports requests, so the debian:13 CI build failed without it --- MANIFEST.in | 1 + README.md | 16 ++++++++++++++++ debian/control | 2 +- debian/install | 1 + docs/configuration.md | 32 +++++++++++++++++++++++++++++++- docs/plugins.md | 7 ++++--- examples/plugins/template.py | 30 ++++++++++++++++++++++++++++++ hpc_eff.spec | 3 +++ 8 files changed, 87 insertions(+), 5 deletions(-) create mode 100755 examples/plugins/template.py diff --git a/MANIFEST.in b/MANIFEST.in index ab93ae2..05e066b 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,3 +1,4 @@ include src/hpc_eff/config.ini.example recursive-include src/hpc_eff/utils *.py *.sql recursive-include src/hpc_eff/pipeline *.py +recursive-include examples *.py diff --git a/README.md b/README.md index 701bbdc..19c73ff 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,22 @@ CPU regulation is mutually exclusive: a node is driven either by carbon/price or by temperature, never both. Details: [docs/regulation-modes.md](docs/regulation-modes.md). +### Plugins + +Every evaluation is a pipeline of plugins: `read` → `score` → `act` → `record`. +Without extra config it behaves exactly as above. To change it, add a +`[PIPELINE]` section, for example CO₂ alone (no price) driving the CPU cap: + +```ini +[PIPELINE] +score = co2, combine +act = cpu_freq +``` + +Your own score or action is any executable that reads JSON on stdin and prints +JSON on stdout. Contract, built-in plugins, security rules and starter files: +[docs/plugins.md](docs/plugins.md). + --- ## Data sources diff --git a/debian/control b/debian/control index 619b713..ba49f36 100644 --- a/debian/control +++ b/debian/control @@ -2,7 +2,7 @@ Source: hpc-eff Section: utils Priority: optional Maintainer: CESNET -Build-Depends: debhelper (>= 12), dh-python, python3-all, python3-setuptools +Build-Depends: debhelper (>= 12), dh-python, python3-all, python3-setuptools, python3-requests Standards-Version: 4.6.0 X-Python3-Version: >= 3.9 diff --git a/debian/install b/debian/install index a84f1b8..45dcd8c 100644 --- a/debian/install +++ b/debian/install @@ -1 +1,2 @@ src/hpc_eff/config.ini.example etc/hpc_eff/ +examples/plugins usr/share/doc/hpc-eff/examples/ diff --git a/docs/configuration.md b/docs/configuration.md index 5a4d245..71b8ad2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -27,6 +27,7 @@ regulator actually reads: | `[TEMPERATURE_SOURCE]` | n/a | yes | | `[CPU_THERMO]` | n/a | yes | | `[GPU_POWER]` | n/a | yes | +| `[PIPELINE]`, `[plugin:NAME]` | optional | optional | --- @@ -158,7 +159,7 @@ weight_co2=0.4 | Key | Default | Meaning | |---|---|---| -| `rating_type` | `price` | `price` \| `average` \| `max`: see [regulation-modes.md](regulation-modes.md#choosing-rating_type) | +| `rating_type` | `price` | `price` \| `co2` \| `average` \| `max`: see [regulation-modes.md](regulation-modes.md#choosing-rating_type) | | `weight_price` | `0.6` | Used by `average` only, normalised against `weight_co2` | | `weight_co2` | `0.4` | Used by `average` only | @@ -310,6 +311,35 @@ constantly. Set both pairs against that range. --- +## `[PIPELINE]` and `[plugin:NAME]`: choosing and extending the steps + +Optional. Replaces the `[MODE] control_mode` presets with an explicit list of +plugins per stage. + +```ini +[PIPELINE] +score = co2, combine +act = cpu_freq + +[plugin:threshold_score] # only for your own executables +exec = /etc/hpc_eff/plugins/threshold_score.py +timeout = 30 +``` + +| Key | Default | Meaning | +|---|---|---| +| `read`, `score`, `act` | empty | Comma-separated plugin names, run in order | +| `record` | `db_log, state_json` | Same, for the persistence stage | +| `[plugin:NAME] exec` | built-in `NAME` | Absolute path to an executable, or `builtin:NAME` | +| `[plugin:NAME] timeout` | `30` | Seconds before the plugin is killed | +| other `[plugin:NAME]` keys | none | Passed to the plugin as-is | + +An invalid pipeline aborts the run with `Config error: ...` before any plugin +starts. Built-in plugin names, the JSON contract and the security rules: +[plugins.md](plugins.md). + +--- + ## Next [regulation-modes.md](regulation-modes.md): what each mode does with these diff --git a/docs/plugins.md b/docs/plugins.md index be14e5c..0c31ab3 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -67,9 +67,10 @@ A `[plugin:NAME]` with `exec` also **replaces a built-in of the same name**, so `[plugin:co2] exec = ...` swaps the data source without touching the rest of the pipeline. `exec = builtin:NAME` selects a built-in explicitly. -Working examples are in [`examples/plugins/`](../examples/plugins): -`threshold_score.py` (score from absolute gCO₂/kWh thresholds) and -`echo_action.py` (an action that changes nothing: a template to copy). +Working examples are in [`examples/plugins/`](../examples/plugins), installed +with the package under `/usr/share/doc/hpc-eff/examples/` (RPM: `/usr/share/doc/hpc_eff/examples/`): +`template.py` (a blank starter), `threshold_score.py` (score from absolute +gCO₂/kWh thresholds) and `echo_action.py` (an action that changes nothing). ### The contract (version 1) diff --git a/examples/plugins/template.py b/examples/plugins/template.py new file mode 100755 index 0000000..ef81291 --- /dev/null +++ b/examples/plugins/template.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +"""Plugin starter. Copy to /etc/hpc_eff/plugins/, chmod 755, owned by root. + +The controller writes one JSON request to stdin and reads one JSON response +from stdout (full contract: docs/plugins.md). Exit non-zero, or send an +"error" string, to report a failure; the evaluation carries on without you. + +Request: {"version": 1, "stage": "...", "plugin": "...", + "config": {}, + "context": {}, + "state": {}} +Response: every field optional. + "set": values for later plugins (and the DB), e.g. {"rating": 6} + "state": values for state.json; "summary" text from several plugins is joined +""" +import json +import sys + +request = json.load(sys.stdin) +options = request["config"] # strings, as written in config.ini +rating = request["context"].get("rating") + +# ... your logic here: score plugins return {"rating": 1..10} under "set", +# action plugins act on `rating` and report under "state". + +print(json.dumps({ + "version": 1, + "set": {}, + "state": {"summary": f"template saw rating {rating}"}, +})) diff --git a/hpc_eff.spec b/hpc_eff.spec index 65154b2..61fcccd 100644 --- a/hpc_eff.spec +++ b/hpc_eff.spec @@ -40,6 +40,8 @@ python3 setup.py build %install python3 setup.py install --root=%{buildroot} --prefix=/usr --skip-build --record=INSTALLED_FILES install -D -m 644 src/hpc_eff/config.ini.example %{buildroot}/etc/hpc_eff/config.ini +install -d %{buildroot}%{_docdir}/%{name}/examples/plugins +install -m 755 examples/plugins/*.py %{buildroot}%{_docdir}/%{name}/examples/plugins/ # The setuptools console_scripts launcher resolves its entry point at runtime # via importlib.metadata. On Python 3.9 that reader does not reliably parse the @@ -62,3 +64,4 @@ chmod 755 %{buildroot}/usr/bin/hpc-eff %defattr(-,root,root,-) %config(noreplace) /etc/hpc_eff/config.ini %doc README.md +%{_docdir}/%{name}/examples From e5abc2566f3851bb1690928736f675693a907fd5 Mon Sep 17 00:00:00 2001 From: pailozka Date: Wed, 7 Oct 2026 14:51:23 +0200 Subject: [PATCH 3/5] fix: add python3-requests to build dependencies in CI workflow --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e7a4b38..4a6fe95 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -16,7 +16,7 @@ jobs: apt-get update apt-get install -y --no-install-recommends \ build-essential devscripts debhelper dh-python \ - python3-all python3-setuptools fakeroot + python3-all python3-setuptools python3-requests fakeroot - name: Build the package run: make deb From 0ef4d3e33a158c0c7b0b19f0afc39a23e25089ec Mon Sep 17 00:00:00 2001 From: pailozka Date: Wed, 7 Oct 2026 15:07:02 +0200 Subject: [PATCH 4/5] fix: declare python3-numpy and python3-requests in Build-Depends and install them in CI via mk-build-deps, so debian/control is the single list --- .github/workflows/ci.yml | 8 +++++--- README.md | 3 ++- debian/control | 2 +- docs/configuration.md | 2 +- docs/plugins.md | 7 ++++--- src/hpc_eff/config.ini.example | 2 +- src/hpc_eff/pipeline/builtins.py | 6 +++++- tests/test_pipeline.py | 11 ++++++++--- 8 files changed, 27 insertions(+), 14 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4a6fe95..e44712c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,11 +12,13 @@ jobs: - uses: actions/checkout@v4 - name: Install build dependencies + # Build-Depends in debian/control is the single list of what the build + # needs; mk-build-deps installs exactly that, so the two cannot drift. run: | apt-get update - apt-get install -y --no-install-recommends \ - build-essential devscripts debhelper dh-python \ - python3-all python3-setuptools python3-requests fakeroot + apt-get install -y --no-install-recommends build-essential devscripts equivs fakeroot + cd /tmp && mk-build-deps --install --remove \ + --tool "apt-get -y --no-install-recommends" "$GITHUB_WORKSPACE/debian/control" - name: Build the package run: make deb diff --git a/README.md b/README.md index 19c73ff..c77cd74 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ killed; they run under a lower clock ceiling while conditions are bad. | [docs/cluster-rollout.md](docs/cluster-rollout.md) | Many nodes, upgrades, uninstall. | | [docs/configuration.md](docs/configuration.md) | Every key of `/etc/hpc_eff/config.ini`. | | [docs/regulation-modes.md](docs/regulation-modes.md) | What each mode does with its input: the rating maths, the thermal bands. | +| [docs/plugins.md](docs/plugins.md) | The score/action pipeline: built-in plugins, writing your own, the JSON contract. | | [docs/monitoring.md](docs/monitoring.md) | Database schema, `state.json`, useful queries. | | [docs/troubleshooting.md](docs/troubleshooting.md) | Symptom-first fault finding. | @@ -133,7 +134,7 @@ Without extra config it behaves exactly as above. To change it, add a ```ini [PIPELINE] -score = co2, combine +score = co2 act = cpu_freq ``` diff --git a/debian/control b/debian/control index ba49f36..b553fa0 100644 --- a/debian/control +++ b/debian/control @@ -2,7 +2,7 @@ Source: hpc-eff Section: utils Priority: optional Maintainer: CESNET -Build-Depends: debhelper (>= 12), dh-python, python3-all, python3-setuptools, python3-requests +Build-Depends: debhelper (>= 12), dh-python, python3-all, python3-setuptools, python3-numpy, python3-requests Standards-Version: 4.6.0 X-Python3-Version: >= 3.9 diff --git a/docs/configuration.md b/docs/configuration.md index 71b8ad2..d21869a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -318,7 +318,7 @@ plugins per stage. ```ini [PIPELINE] -score = co2, combine +score = co2 act = cpu_freq [plugin:threshold_score] # only for your own executables diff --git a/docs/plugins.md b/docs/plugins.md index 0c31ab3..d173e5c 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -32,12 +32,13 @@ omitted. A stage you leave out is empty, except `record`, which defaults to ```ini # Wattnet (or Nowtricity) CO2 alone drives the CPU cap. No price fetch. [PIPELINE] -score = co2, combine +score = co2 act = cpu_freq ``` -`combine` with no `price` plugin simply passes the CO₂ rating through. To pick -the data source, set `[CO2_API] TYPE = wattnet`. +On its own, `co2` makes its grade the `rating`. Add `combine` only when `price` +is listed too, to choose how the two mix (`[aggregation]`). To pick the data +source, set `[CO2_API] TYPE = wattnet`. An invalid pipeline (unknown key, unknown plugin, a plugin listed under the wrong stage, a bad executable) aborts the run before anything starts, with diff --git a/src/hpc_eff/config.ini.example b/src/hpc_eff/config.ini.example index a5fbcaf..d569762 100644 --- a/src/hpc_eff/config.ini.example +++ b/src/hpc_eff/config.ini.example @@ -157,7 +157,7 @@ weight_co2=0.4 # Example: regulate by CO2 alone (Nowtricity or Wattnet per [CO2_API]), no price: # #[PIPELINE] -#score = co2, combine +#score = co2 #act = cpu_freq # # Example: swap in your own executable. A [plugin:NAME] section with `exec` diff --git a/src/hpc_eff/pipeline/builtins.py b/src/hpc_eff/pipeline/builtins.py index fe43d0c..2ee5f44 100644 --- a/src/hpc_eff/pipeline/builtins.py +++ b/src/hpc_eff/pipeline/builtins.py @@ -165,7 +165,11 @@ def co2(req, config, rt): # CO2 grade (1 low - 10 high) doubles as the CO2 rating "rating_co2": grade if isinstance(grade, int) else None, } - return {"set": values, "state": dict(values)} + state = dict(values) + if values["rating_co2"] is not None: + # On its own, the CO2 grade is the rating. `combine`, when listed, overrides it. + values["rating"] = state["rating"] = values["rating_co2"] + return {"set": values, "state": state} def combine(req, config, rt): diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 2a8a477..a754d96 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -182,7 +182,7 @@ def row(self): def test_co2_only_pipeline_fetches_no_price(self): with mock.patch.object(pb, "get_current_energy_price", side_effect=AssertionError("price fetched")): - self.run_cfg("[PIPELINE]\nscore = co2, combine\nact = cpu_freq\n") + self.run_cfg("[PIPELINE]\nscore = co2\nact = cpu_freq\n") self.set_freq.assert_called_once() self.assertEqual(self.set_freq.call_args[0][0], 7) # rating = CO2 grade entry = self.entry() @@ -194,10 +194,15 @@ def test_co2_only_pipeline_fetches_no_price(self): self.assertEqual(self.row()["rating"], 7) self.assertEqual(self.row()["freq_max"], 2200000) static = json.loads(self.state.read_text())["static"] - self.assertEqual(static["plugins"]["score"], ["co2", "combine"]) + self.assertEqual(static["plugins"]["score"], ["co2"]) + + def test_combine_after_co2_still_applies_aggregation(self): + with mock.patch.object(pb, "get_current_energy_price", side_effect=AssertionError("price fetched")): + self.run_cfg("[aggregation]\nrating_type=max\n[PIPELINE]\nscore = co2, combine\nact = cpu_freq\n") + self.assertEqual(self.set_freq.call_args[0][0], 7) # no price plugin: CO2 passes through def test_co2_failure_falls_back_to_neutral_rating(self): - self.run_cfg("[PIPELINE]\nscore = co2, combine\nact = cpu_freq\n", co2=RuntimeError("401")) + self.run_cfg("[PIPELINE]\nscore = co2\nact = cpu_freq\n", co2=RuntimeError("401")) self.assertEqual(self.set_freq.call_args[0][0], 5) self.assertEqual(self.entry()["co2_grade"], "unknown") From 56a08d649bfa3f1c7054633b90c2a343f192fe4a Mon Sep 17 00:00:00 2001 From: pailozka Date: Tue, 13 Oct 2026 10:20:53 +0200 Subject: [PATCH 5/5] docs: plugin install steps, cpu_freq explained, several actions per stage --- docs/plugins.md | 94 ++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 90 insertions(+), 4 deletions(-) diff --git a/docs/plugins.md b/docs/plugins.md index d173e5c..1d53f0e 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -73,6 +73,55 @@ with the package under `/usr/share/doc/hpc-eff/examples/` (RPM: `/usr/share/doc/ `template.py` (a blank starter), `threshold_score.py` (score from absolute gCO₂/kWh thresholds) and `echo_action.py` (an action that changes nothing). +### Installing a plugin, step by step + +1. **Put the executable in a root-owned directory.** The file and its + directory must both be owned by root and not writable by group or others, + or the run aborts (see [Security](#security)). + + ```bash + sudo mkdir -p /etc/hpc_eff/plugins + sudo cp my_score.py /etc/hpc_eff/plugins/ + sudo chown -R root:root /etc/hpc_eff/plugins + sudo chmod 755 /etc/hpc_eff/plugins /etc/hpc_eff/plugins/my_score.py + ``` + + The first line of the script must be a shebang such as `#!/usr/bin/env python3`. + +2. **Declare it in `/etc/hpc_eff/config.ini`** (the same file as `[PIPELINE]`; + back it up first). The name after `plugin:` is the name you list in the + stage: + + ```ini + [PIPELINE] + score = co2, my_score + act = cpu_freq + + [plugin:my_score] + exec = /etc/hpc_eff/plugins/my_score.py + ``` + +3. **Test it by hand** before the next cron run. A plugin is a plain program, + so you can feed it a request yourself: + + ```bash + echo '{"version":1,"stage":"score","plugin":"my_score","config":{},"context":{"co2_current":300},"state":{}}' \ + | sudo /etc/hpc_eff/plugins/my_score.py + ``` + + It should print one JSON object. To try the whole pipeline without + touching the CPU, temporarily set `act = echo_action` (it only reports). + +4. **Run once and look at the result.** With `DEBUG=yes` in `[SYSTEM]`, run + `sudo hpc-eff`, then read `state_json_path` (default + `/var/lib/hpc_eff/state.json`): your `summary` appears under `current`, and + any failure is listed under `plugin_errors`. + +An error such as `Config error: ...` means the pipeline did not start and +nothing was changed. Typical causes: the plugin name has no `[plugin:NAME]` +section, it is listed under the wrong stage, or the file or its directory is +not root-owned. + ### The contract (version 1) The controller starts the executable, writes **one JSON request** to its stdin @@ -141,11 +190,28 @@ Put plugins in a root-owned directory such as `/etc/hpc_eff/plugins/` `[plugin:NAME]` section, reach the plugin on stdin, never on the command line. Keep `config.ini` readable by root only if it holds credentials. -### Two actions, one knob +### Several actions, one knob each + +A stage takes a comma-separated list and runs it in the order written, so +`act` can hold more than one plugin: + +```ini +act = cpu_freq, echo_action +``` + +Here `cpu_freq` caps the CPU first, then `echo_action` reports. The rule is +**one action per resource**: if two actions set the same thing, the later one +overwrites the earlier one. -Two `act` plugins can set the same knob (for example `cpu_freq` and -`cpu_thermo` both set the CPU max frequency). The later one wins, and a debug -line names both whenever two actions report a `freq_max`. Pick one per resource. +| Combination | Result | +|---|---| +| `cpu_freq, echo_action` | fine: `echo_action` changes nothing | +| `cpu_freq, gpu_power` | fine: they change different hardware (CPU and GPU) | +| `cpu_freq, cpu_thermo` | conflict: both set the CPU max frequency, the last one wins | + +Whenever two actions report a `freq_max`, a debug line names both. The same +order rule applies to `score`: the last plugin that sets `rating` wins, which +is how a plugin listed after `combine` can override the built-in rating. ## Built-in plugins @@ -162,6 +228,26 @@ line names both whenever two actions report a `freq_max`. Pick one per resource. | `db_log` | record | the whole context | one row in `hpc_eff_log` | | `state_json` | record | this run's entry | `state.json` | +### What `act = cpu_freq` does + +The `score` stage only produces a number, `rating`. Nothing on the machine +changes until an `act` plugin uses it, and `cpu_freq` is the built-in that +does: + +1. It reads `rating` (1 to 10; 5 if no score plugin set one). +2. It looks up a maximum frequency for that rating in the list for your CPU + model in `[frequency_tables]`. Rating 1 gets the highest entry, rating 10 + the lowest, and ratings in between are spread across the list. +3. It sets the CPU's **maximum** frequency to that value with + `cpupower frequency-set`. The CPU can still clock down below the cap; it + just cannot go above it. +4. It reports the chosen value as `selected_freq_khz` (stored as `freq_max`). + +So a dirty or expensive hour (high rating) caps the CPU low, and a clean or +cheap hour (low rating) lets it run at full speed. Without any `act` plugin +the scores are still computed and logged, but the CPU is left alone. Setting +the cap needs root. + `[aggregation] rating_type` accepts `price`, `co2`, `average` and `max`. `co2` uses the CO₂ rating alone and falls back to the price rating, then to 5, when CO₂ data is missing.