feat: support custom CUR label aliases

.seqera/context/PIPELINE.md

@@ -51,21 +51,22 @@ Input CSV (id, workspace, group, logs, platform, token_env)
 | Process                           | Container                       | Input                                       | Output                      |
 | --------------------------------- | ------------------------------- | ------------------------------------------- | --------------------------- |
 | `EXTRACT_TARBALL`                 | ubuntu:22.04                    | `(meta, tarball)`                           | `(meta, dir)` of JSON files |
-| `NORMALIZE_BENCHMARK_JSONL`       | wave python/duckdb/jinja2/typer | `data_dir`, `cur_parquet`                   | `jsonl_bundle/`             |
+| `NORMALIZE_BENCHMARK_JSONL`       | wave python/duckdb/jinja2/typer | `data_dir`, `cur_parquet`, `cur_label_map`  | `jsonl_bundle/`             |
 | `AGGREGATE_BENCHMARK_REPORT_DATA` | wave python/duckdb/jinja2/typer | `jsonl_bundle/`                             | `report_data.json`          |
 | `RENDER_BENCHMARK_REPORT`         | wave python/duckdb/jinja2/typer | `report_data.json`, `brand.yml`, `logo.svg` | `benchmark_report.html`     |
 
 ## Key Parameters
 
-| Parameter                   | Default                       | Purpose                                     |
-| --------------------------- | ----------------------------- | ------------------------------------------- |
-| `input`                     | required                      | CSV samplesheet of run IDs / external paths |
-| `outdir`                    | `results`                     | Output directory                            |
-| `generate_benchmark_report` | `false`                       | Enable the benchmark pipeline               |
-| `benchmark_aws_cur_report`  | `null`                        | AWS CUR parquet for cost analysis           |
-| `seqera_api_endpoint`       | `https://api.cloud.seqera.io` | Platform API base URL                       |
-| `java_truststore_path`      | `null`                        | Custom Java truststore for private certs    |
-| `java_truststore_password`  | `null`                        | Truststore password                         |
+| Parameter                     | Default                       | Purpose                                         |
+| ----------------------------- | ----------------------------- | ----------------------------------------------- |
+| `input`                       | required                      | CSV samplesheet of run IDs / external paths     |
+| `outdir`                      | `results`                     | Output directory                                |
+| `generate_benchmark_report`   | `false`                       | Enable the benchmark pipeline                   |
+| `benchmark_aws_cur_report`    | `null`                        | AWS CUR parquet for cost analysis               |
+| `benchmark_aws_cur_label_map` | `null`                        | Optional YAML alias map for CUR resource labels |
+| `seqera_api_endpoint`         | `https://api.cloud.seqera.io` | Platform API base URL                           |
+| `java_truststore_path`        | `null`                        | Custom Java truststore for private certs        |
+| `java_truststore_password`    | `null`                        | Truststore password                             |
 
 ## Input Schema
 

.seqera/skills/rebuild-benchmark-report-locally/SKILL.md

@@ -37,6 +37,7 @@ Use those boundaries to isolate failures instead of re-running everything.
 uv run --with typer --with pyyaml --with pyarrow \
   python bin/benchmark_report.py normalize-jsonl \
   --data-dir /path/to/json_data \
+  --cost-label-map /path/to/cur_label_map.yml \
   --output-dir /tmp/jsonl_bundle
 ```
 
@@ -47,6 +48,7 @@ uv run --with typer --with pyyaml --with pyarrow \
   python bin/benchmark_report.py normalize-jsonl \
   --data-dir /path/to/json_data \
   --costs /path/to/cur.parquet \
+  --cost-label-map /path/to/cur_label_map.yml \
   --output-dir /tmp/jsonl_bundle
 ```
 

AGENTS.md

@@ -17,9 +17,10 @@ input CSV (id, workspace, group, logs, fusion)
 
 | Param                       | Default                       | Purpose                           |
 | --------------------------- | ----------------------------- | --------------------------------- |
-| `generate_benchmark_report` | false                         | Enable benchmark report           |
-| `benchmark_aws_cur_report`  | null                          | AWS CUR parquet for cost analysis |
-| `seqera_api_endpoint`       | `https://api.cloud.seqera.io` | Platform API URL                  |
+| `generate_benchmark_report`   | false                         | Enable benchmark report                           |
+| `benchmark_aws_cur_report`    | null                          | AWS CUR parquet for cost analysis                 |
+| `benchmark_aws_cur_label_map` | null                          | YAML aliases for custom CUR resource label names  |
+| `seqera_api_endpoint`         | `https://api.cloud.seqera.io` | Platform API URL                                  |
 
 ## Plugins
 
@@ -38,6 +39,13 @@ uv run --with typer --with pyyaml --with pyarrow \
   python bin/benchmark_report.py normalize-jsonl \
   --data-dir /path/to/json_data --output-dir /tmp/jsonl_bundle
 
+# Normalize raw run JSON with CUR cost enrichment + custom label aliases:
+uv run --with typer --with pyyaml --with pyarrow \
+  python bin/benchmark_report.py normalize-jsonl \
+  --data-dir /path/to/json_data --costs /path/to/cur.parquet \
+  --cost-label-map /path/to/cur_label_map.yml \
+  --output-dir /tmp/jsonl_bundle
+
 # Aggregate JSONL bundle to report data:
 uv run --with typer --with pyyaml \
   python bin/benchmark_report.py aggregate-report-data \

README.md

@@ -100,8 +100,56 @@ nextflow run seqeralabs/nf-aggregate \
     --benchmark_aws_cur_report ./aws_cost_report.parquet
 ```
 
+If your CUR export uses custom resource label names, pass an optional YAML alias map with `benchmark_aws_cur_label_map`:
+
+```
+nextflow run seqeralabs/nf-aggregate \
+    --input run_ids.csv \
+    --outdir ./results \
+    --generate_benchmark_report \
+    --benchmark_aws_cur_report ./aws_cost_report.parquet \
+    --benchmark_aws_cur_label_map ./cur_label_map.yml
+```
+
 The benchmark report can be generated without cost data - simply omit the `--benchmark_aws_cur_report` parameter if cost analysis is not needed.
 
+When CUR data is provided, nf-aggregate only joins cost rows that carry the resource labels needed to match a Nextflow task back to a benchmark run. By default the logical fields map to:
+
+- run ID: `user_unique_run_id` or `user_nf_unique_run_id`
+- process name: `user_pipeline_process`
+- task hash: `user_task_hash`
+
+To accept manual/custom label names, create a YAML file listing aliases for the logical fields:
+
+```
+run_id:
+  - my_team_run_id
+  - user_unique_run_id
+process:
+  - my_process_label
+  - user_pipeline_process
+task_hash:
+  - my_task_hash_label
+  - user_task_hash
+```
+
+Aliases are tried in order, then the built-in defaults are still checked as a fallback.
+
+The normalizer accepts both of the common CUR layouts:
+
+- flattened CUR columns such as `resource_tags_user_unique_run_id`, `resource_tags_my_process_label`, and `resource_tags_user_task_hash`
+- map-style `resource_tags` entries containing `user_unique_run_id`, `my_process_label`, and `user_task_hash`
+
+If those labels are missing from the CUR export, the benchmark report still renders, but CUR-backed cost rows cannot be associated with runs or tasks.
+
+For AWS Batch or other cloud executors that propagate resource labels into CUR tags, configure labels equivalent to:
+
+```
+user_unique_run_id=${workflow.runId}
+user_pipeline_process=${task.process}
+user_task_hash=${task.hash}
+```
+
 For a checked-in real-world example that exercises external run JSON directories plus a tiny filtered cost parquet, see:
 
 - `workflows/nf_aggregate/assets/test_benchmark_realworld_costs.csv`
@@ -118,6 +166,8 @@ python scripts/build_filtered_cost_sidecar.py \
     --output workflows/nf_aggregate/assets/test_benchmark_realworld_costs.parquet
 ```
 
+The helper script also accepts `--cost-label-map ./cur_label_map.yml` when the monthly CUR export uses custom run-id label aliases.
+
 Add `--include-red-herring` only if you want one synthetic non-benchmark row for robustness testing.
 
 ## Output

bin/benchmark_report.py

@@ -28,12 +28,19 @@ def normalize_jsonl_cmd(
     data_dir: Path = typer.Option(..., exists=True, help="Directory containing run JSON files"),
     output_dir: Path = typer.Option(Path("jsonl_bundle"), help="Output JSONL bundle directory"),
     costs: Path = typer.Option(None, help="Optional AWS CUR parquet file"),
+    cost_label_map: Path = typer.Option(None, help="Optional YAML file mapping logical CUR labels to custom tag aliases"),
     machines_dir: Path = typer.Option(None, help="Optional machine metrics CSV directory"),
 ) -> None:
     """Normalize raw run JSON into runs/tasks/metrics JSONL files."""
     from benchmark_report_normalize import normalize_jsonl
 
-    normalize_jsonl(data_dir=data_dir, output_dir=output_dir, costs_parquet=costs, machines_dir=machines_dir)
+    normalize_jsonl(
+        data_dir=data_dir,
+        output_dir=output_dir,
+        costs_parquet=costs,
+        cost_label_map=cost_label_map,
+        machines_dir=machines_dir,
+    )
 
 
 @app.command("aggregate-report-data")

bin/normalize_benchmark_jsonl.py

@@ -14,9 +14,16 @@ def main(argv: list[str] | None = None) -> None:
     parser.add_argument("--data-dir", type=Path, required=True, help="Directory containing run JSON files")
     parser.add_argument("--output-dir", type=Path, default=Path("jsonl_bundle"), help="Output JSONL bundle directory")
     parser.add_argument("--costs", type=Path, default=None, help="Optional AWS CUR parquet file")
+    parser.add_argument("--cost-label-map", type=Path, default=None, help="Optional YAML mapping for CUR resource label aliases")
     parser.add_argument("--machines-dir", type=Path, default=None, help="Directory containing machine metrics CSVs")
     args = parser.parse_args(argv)
-    normalize_jsonl(data_dir=args.data_dir, output_dir=args.output_dir, costs_parquet=args.costs, machines_dir=args.machines_dir)
+    normalize_jsonl(
+        data_dir=args.data_dir,
+        output_dir=args.output_dir,
+        costs_parquet=args.costs,
+        cost_label_map=args.cost_label_map,
+        machines_dir=args.machines_dir,
+    )
 
 
 if __name__ == "__main__":

bin/test_build_filtered_cost_sidecar.py

@@ -0,0 +1,25 @@
+from pathlib import Path
+import sys
+
+import pyarrow as pa
+
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import build_filtered_cost_sidecar as sidecar
+
+
+def test_build_run_ids_prefers_custom_map_alias_over_default_flat_column():
+    aliases = sidecar._dedupe_aliases(["custom_run", *sidecar.DEFAULT_RUN_ID_ALIASES])
+    table = pa.table(
+        {
+            "resource_tags_user_unique_run_id": ["run-default"],
+            "resource_tags": [[("custom_run", "run-custom")]],
+        }
+    )
+
+    run_id_column = sidecar.choose_run_id_column(table.schema, aliases)
+
+    assert run_id_column == "resource_tags_user_unique_run_id"
+    assert sidecar.build_run_ids(table, run_id_column, aliases) == ["run-custom"]

docs/DESIGN.md

@@ -33,6 +33,7 @@ Command:
 
 ```bash
 python bin/benchmark_report.py normalize-jsonl --data-dir <run_json_dir> --output-dir <jsonl_bundle>
+python bin/benchmark_report.py normalize-jsonl --data-dir <run_json_dir> --costs <cur.parquet> --cost-label-map <cur_label_map.yml> --output-dir <jsonl_bundle>
 ```
 
 Responsibilities:
@@ -120,6 +121,13 @@ uv run --with typer --with pyyaml --with pyarrow \
   python bin/benchmark_report.py normalize-jsonl \
   --data-dir /path/to/json_data --output-dir /tmp/jsonl_bundle
 
+uv run --with typer --with pyyaml --with pyarrow \
+  python bin/benchmark_report.py normalize-jsonl \
+  --data-dir /path/to/json_data \
+  --costs /path/to/cur.parquet \
+  --cost-label-map /path/to/cur_label_map.yml \
+  --output-dir /tmp/jsonl_bundle
+
 uv run --with typer --with pyyaml \
   python bin/benchmark_report.py aggregate-report-data \
   --jsonl-dir /tmp/jsonl_bundle --output /tmp/report_data.json

modules/local/normalize_benchmark_jsonl/bin/benchmark_report_normalize.py

@@ -11,6 +11,12 @@
 
 import typer
 
+DEFAULT_COST_LABEL_ALIASES: dict[str, list[str]] = {
+    "run_id": ["user_unique_run_id", "user_nf_unique_run_id"],
+    "process": ["user_pipeline_process"],
+    "task_hash": ["user_task_hash"],
+}
+
 
 def _run_group(run: dict[str, Any]) -> str:
     return run["meta"]["group"]
@@ -239,22 +245,78 @@ def _iter_parquet_rows(costs_parquet: Path):
             yield cols, row
 
 
-def _normalize_cost_rows(costs_parquet: Path) -> list[dict[str, Any]]:
+def _dedupe_aliases(values: list[str]) -> list[str]:
+    seen: set[str] = set()
+    aliases: list[str] = []
+    for value in values:
+        alias = str(value).strip()
+        if not alias or alias in seen:
+            continue
+        seen.add(alias)
+        aliases.append(alias)
+    return aliases
+
+
+def _normalise_label_aliases(value: Any, field: str) -> list[str]:
+    if value is None:
+        return []
+    if isinstance(value, str):
+        return _dedupe_aliases([value])
+    if isinstance(value, list):
+        if not all(isinstance(item, str) for item in value):
+            raise ValueError(f"benchmark_aws_cur_label_map field '{field}' must contain only strings")
+        return _dedupe_aliases(value)
+    raise ValueError(f"benchmark_aws_cur_label_map field '{field}' must be a string or list of strings")
+
+
+def _load_cost_label_aliases(cost_label_map: Path | None = None) -> dict[str, list[str]]:
+    aliases = {field: list(defaults) for field, defaults in DEFAULT_COST_LABEL_ALIASES.items()}
+    if cost_label_map is None or cost_label_map.name in {"NO_FILE", "NO_FILE_CUR_LABEL_MAP"}:
+        return aliases
+
+    try:
+        import yaml
+    except ImportError as exc:
+        raise RuntimeError("pyyaml is required to read benchmark_aws_cur_label_map") from exc
+
+    with cost_label_map.open() as handle:
+        raw_config = yaml.safe_load(handle) or {}
+
+    if not isinstance(raw_config, dict):
+        raise ValueError("benchmark_aws_cur_label_map must contain a YAML mapping")
+
+    unknown_fields = set(raw_config) - set(DEFAULT_COST_LABEL_ALIASES)
+    if unknown_fields:
+        unknown_fields_csv = ", ".join(sorted(unknown_fields))
+        raise ValueError(f"benchmark_aws_cur_label_map contains unsupported fields: {unknown_fields_csv}")
+
+    for field, defaults in DEFAULT_COST_LABEL_ALIASES.items():
+        user_aliases = _normalise_label_aliases(raw_config.get(field), field)
+        aliases[field] = _dedupe_aliases(user_aliases + defaults)
+
+    return aliases
+
+
+def _resolve_cost_label_value(row: dict[str, Any], tags: dict[str, Any], aliases: list[str]) -> Any:
+    for alias in aliases:
+        flat_value = row.get(f"resource_tags_{alias}")
+        if flat_value not in (None, ""):
+            return flat_value
+        tag_value = tags.get(alias)
+        if tag_value not in (None, ""):
+            return tag_value
+    return None
+
+
+def _normalize_cost_rows(costs_parquet: Path, cost_label_map: Path | None = None) -> list[dict[str, Any]]:
     grouped: dict[tuple[str, str, str], dict[str, float | str]] = {}
-    is_map: bool | None = None
+    label_aliases = _load_cost_label_aliases(cost_label_map)
 
     for cols, row in _iter_parquet_rows(costs_parquet):
-        if is_map is None:
-            is_map = "resource_tags" in cols and "resource_tags_user_unique_run_id" not in cols
-        if is_map:
-            tags = _to_tags_dict(row.get("resource_tags"))
-            run_id = tags.get("user_unique_run_id") or tags.get("user_nf_unique_run_id")
-            process = tags.get("user_pipeline_process")
-            hash_val = tags.get("user_task_hash")
-        else:
-            run_id = row.get("resource_tags_user_unique_run_id") or row.get("resource_tags_user_nf_unique_run_id")
-            process = row.get("resource_tags_user_pipeline_process")
-            hash_val = row.get("resource_tags_user_task_hash")
+        tags = _to_tags_dict(row.get("resource_tags")) if "resource_tags" in cols else {}
+        run_id = _resolve_cost_label_value(row, tags, label_aliases["run_id"])
+        process = _resolve_cost_label_value(row, tags, label_aliases["process"])
+        hash_val = _resolve_cost_label_value(row, tags, label_aliases["task_hash"])
 
         if not run_id:
             continue
@@ -434,7 +496,13 @@ def _summarise_machines(machines_dir: Path) -> list[dict[str, Any]]:
     return results
 
 
-def normalize_jsonl(data_dir: Path, output_dir: Path, costs_parquet: Path | None = None, machines_dir: Path | None = None) -> None:
+def normalize_jsonl(
+    data_dir: Path,
+    output_dir: Path,
+    costs_parquet: Path | None = None,
+    machines_dir: Path | None = None,
+    cost_label_map: Path | None = None,
+) -> None:
     runs = load_run_data(data_dir)
     if not runs:
         typer.echo("No run data found", err=True)
@@ -451,7 +519,7 @@ def normalize_jsonl(data_dir: Path, output_dir: Path, costs_parquet: Path | None
     _write_jsonl(output_dir / "metrics.jsonl", metric_rows)
 
     if costs_parquet and costs_parquet.exists() and costs_parquet.name != "NO_FILE":
-        cost_rows = _normalize_cost_rows(costs_parquet)
+        cost_rows = _normalize_cost_rows(costs_parquet, cost_label_map=cost_label_map)
         _write_jsonl(output_dir / "costs.jsonl", cost_rows)
 
     if machines_dir and machines_dir.exists() and any(machines_dir.glob("*.csv")):

modules/local/normalize_benchmark_jsonl/main.nf

@@ -6,6 +6,7 @@ process NORMALIZE_BENCHMARK_JSONL {
     input:
     path data_dir
     path benchmark_aws_cur_report
+    path benchmark_aws_cur_label_map
     path machines_dir
 
     output:
@@ -14,11 +15,13 @@ process NORMALIZE_BENCHMARK_JSONL {
 
     script:
     def cost_flag = benchmark_aws_cur_report.name != 'NO_FILE' && benchmark_aws_cur_report.name != 'NO_FILE_CUR' ? "--costs ${benchmark_aws_cur_report}" : ""
+    def label_map_flag = benchmark_aws_cur_label_map.name != 'NO_FILE' && benchmark_aws_cur_label_map.name != 'NO_FILE_CUR_LABEL_MAP' ? "--cost-label-map ${benchmark_aws_cur_label_map}" : ""
     def machines_flag = machines_dir.name != 'NO_FILE' && machines_dir.name != 'NO_FILE_MACHINES' ? "--machines-dir ${machines_dir}" : ""
     """
     normalize_benchmark_jsonl.py \\
         --data-dir ${data_dir} \\
         ${cost_flag} \\
+        ${label_map_flag} \\
         ${machines_flag} \\
         --output-dir jsonl_bundle