Skip to content

Commit 5807555

Browse files
edmundmilleredmundmiller
committed
refactor: wire up 4-step benchmark pipeline in workflow
Replace monolithic BENCHMARK_REPORT process with the decomposed pipeline: API (head job) → CLEAN_JSON → BUILD_TABLES → RENDER_REPORT CLEAN_CUR ↗ (optional) - Step 0: API fetching unchanged (head job, nf-boost map{}) - Step 1: CLEAN_JSON — raw JSON → normalized CSVs - Step 2: CLEAN_CUR — AWS CUR parquet → costs CSV (optional) - Step 3: BUILD_TABLES — DuckDB queries → result JSONs - Step 4: RENDER_REPORT — pre-computed data → HTML Update docs/DESIGN.md with new architecture diagram and local testing instructions. Update AGENTS.md rebuild commands.
1 parent 24253f6 commit 5807555

3 files changed

Lines changed: 207 additions & 75 deletions

File tree

‎AGENTS.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,16 @@ Additional paths (always active):
3737
## Rebuild Command (local testing)
3838

3939
```bash
40+
# New decomposed pipeline:
41+
uv run --with duckdb --with typer --with pyyaml python bin/clean_json.py \
42+
--data-dir modules/local/benchmark_report/tests/data --output-dir /tmp/cleaned
43+
uv run --with duckdb --with typer python bin/build_tables.py \
44+
--runs-csv /tmp/cleaned/runs.csv --tasks-csv /tmp/cleaned/tasks.csv \
45+
--metrics-csv /tmp/cleaned/metrics.csv --output-dir /tmp/tables
46+
uv run --with jinja2 --with typer --with pyyaml python bin/render_report.py \
47+
--tables-dir /tmp/tables --brand assets/brand.yml --output /tmp/benchmark_report.html
48+
49+
# Legacy monolithic (still works):
4050
uv run --with duckdb --with jinja2 --with typer --with pyyaml --with pyarrow \
4151
python bin/benchmark_report.py \
4252
--data-dir modules/local/benchmark_report/tests/data \

‎docs/DESIGN.md‎

Lines changed: 164 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -2,90 +2,118 @@
22

33
## Overview
44

5-
nf-boost `request()` + `map` → DuckDB → eCharts. Zero containers for data fetch, one lightweight Python container for report generation.
5+
nf-boost `request()` + `map` → DuckDB → eCharts. Zero containers for data fetch, multi-step pipeline for report generation.
66

77
## Architecture
88

99
```
1010
Nextflow Pipeline (nf-boost)
11-
┌────────────────────────────────────────────────────────┐
12-
│ │
13-
│ input CSV ──→ .map { fetchWorkflow(it) } │
14-
│ ├─ GET /workflow/{id} │
15-
│ ├─ GET /workflow/{id}/metrics │
16-
│ ├─ GET /workflow/{id}/tasks │
17-
│ └─ GET /workflow/{id}/progress │
18-
│ │
19-
│ ↓ channel: [meta, workflow_json, tasks_json, │
20-
│ metrics_json, progress_json] │
21-
│ │
22-
│ .collect() ──→ BENCHMARK_REPORT (Python + DuckDB) │
23-
│ + optional AWS CUR parquet │
24-
│ ↓ │
25-
│ benchmark_report.html (eCharts) │
26-
└────────────────────────────────────────────────────────┘
11+
┌────────────────────────────────────────────────────────────┐
12+
│ │
13+
│ input CSV ──→ .map { fetchWorkflow(it) } (head job) │
14+
│ ├─ GET /workflow/{id} │
15+
│ ├─ GET /workflow/{id}/metrics │
16+
│ ├─ GET /workflow/{id}/tasks │
17+
│ └─ GET /workflow/{id}/progress │
18+
│ │
19+
│ ↓ collect JSON files into directory │
20+
│ │
21+
│ ┌─ CLEAN_JSON ──→ runs.csv, tasks.csv, metrics.csv │
22+
│ │ │
23+
│ ├─ CLEAN_CUR ───→ costs.csv (optional, CUR 1.0 or 2.0) │
24+
│ │ │
25+
│ ├─ BUILD_TABLES ─→ query result JSONs (9 files) │
26+
│ │ (joins, aggregation, DuckDB queries) │
27+
│ │ │
28+
│ └─ RENDER_REPORT → benchmark_report.html (eCharts) │
29+
│ (strictly HTML rendering, no queries) │
30+
└────────────────────────────────────────────────────────────┘
2731
```
2832

33+
Additional paths (always active):
34+
- `SEQERA_RUNS_DUMP` (tower-cli) → run dump dirs → `MULTIQC` + `PLOT_RUN_GANTT`
35+
2936
## Data Flow
3037

31-
### Step 1: Fetch via nf-boost `request()` + `map`
38+
### Step 0: Fetch via nf-boost `request()` + `map` (head job)
3239

33-
No process needed. Pure Nextflow `map` operator calls the Seqera API directly:
40+
No process needed. Pure Nextflow `map` operator calls the Seqera API directly
41+
via `SeqeraApi.fetchRunData()`. JSON files collected into a directory.
3442

35-
```nextflow
36-
include { request; fromJson; toJson } from 'plugin/nf-boost'
43+
### Step 1: CLEAN_JSON — Normalize raw JSON → CSVs
3744

38-
def fetchRun(meta, apiEndpoint) {
39-
def token = System.getenv("TOWER_ACCESS_TOKEN")
40-
def headers = ["Authorization": "Bearer ${token}"]
41-
def wsId = resolveWorkspaceId(meta.workspace, apiEndpoint, headers)
45+
Script: `bin/clean_json.py`
4246

43-
def workflow = apiGet("${apiEndpoint}/workflow/${meta.id}?workspaceId=${wsId}", headers)
44-
def metrics = apiGet("${apiEndpoint}/workflow/${meta.id}/metrics?workspaceId=${wsId}", headers)
45-
def tasks = apiGetPaginated("${apiEndpoint}/workflow/${meta.id}/tasks?workspaceId=${wsId}", headers)
46-
def progress = apiGet("${apiEndpoint}/workflow/${meta.id}/progress?workspaceId=${wsId}", headers)
47+
Reads run JSON files and produces:
48+
- `runs.csv` — one row per workflow run (includes `cached` count)
49+
- `tasks.csv` — one row per task (filtered: keeps COMPLETED + CACHED, drops FAILED)
50+
- `metrics.csv` — one row per process metric field
4751

48-
return [meta, workflow, metrics, tasks, progress]
49-
}
50-
```
52+
### Step 2: CLEAN_CUR — Normalize AWS CUR parquet → CSV (optional)
53+
54+
Script: `bin/clean_cur.py`
55+
56+
Auto-detects CUR format:
57+
- **CUR 2.0** (MAP format): `resource_tags` is `MAP(VARCHAR, VARCHAR)`
58+
- **CUR 1.0** (flattened): `resource_tags_user_unique_run_id`, etc.
59+
60+
Produces: `costs.csv` with columns: `run_id, process, hash, cost, used_cost, unused_cost`
61+
62+
### Step 3: BUILD_TABLES — DuckDB joins/aggregation → query result JSONs
63+
64+
Script: `bin/build_tables.py`
65+
66+
Reads CSVs, runs DuckDB queries, outputs JSON files:
67+
- `benchmark_overview.json` — Pipeline × group matrix
68+
- `run_summary.json` — Infrastructure settings (includes `cachedCount`)
69+
- `run_metrics.json` — Duration, CPU time, efficiency
70+
- `run_costs.json` — Per-run costs (task-level + optional CUR)
71+
- `process_stats.json` — Per-process mean ± SD
72+
- `task_instance_usage.json` — Instance type counts
73+
- `task_table.json` — Full task table
74+
- `task_scatter.json` — Realtime vs staging scatter data
75+
- `cost_overview.json` — CUR cost breakdown (if available)
5176

52-
### Step 2: Write JSON → DuckDB in Python process
77+
### Step 4: RENDER_REPORT — HTML rendering (no queries)
5378

54-
The `BENCHMARK_REPORT` process receives all JSON data, loads into DuckDB, joins with optional AWS CUR parquet, and renders eCharts HTML.
79+
Script: `bin/render_report.py`
5580

56-
#### DuckDB Tables
81+
Loads pre-computed JSON files, renders self-contained HTML with eCharts.
82+
**Does no DuckDB queries.** Strictly presentation layer.
83+
84+
Report sections:
85+
1. **Benchmark Overview** — pipeline × group matrix
86+
2. **Run Overview** — summary table + metrics charts
87+
3. **Run Metrics** — wall time, CPU time, cost, status, efficiency, I/O
88+
4. **Workflow Status** — succeeded/failed/cached stacked bars
89+
5. **Process Overview** — dot + error bar charts, cost per process
90+
6. **Task Overview** — instance usage, scatter, box plots, task table
91+
92+
## DuckDB Tables
5793

5894
**`runs`** — one row per workflow run:
59-
- run_id, group, pipeline, run_name, status, start, complete, duration
60-
- cpu_efficiency, memory_efficiency, cpu_time, read_bytes, write_bytes
61-
- succeeded, failed, cached, fusion_enabled, wave_enabled
95+
- run_id, group, pipeline, run_name, status, start, complete, duration_ms
96+
- succeeded, failed, **cached** (from `workflow.stats.cachedCount`)
97+
- cpu_efficiency, memory_efficiency, cpu_time_ms, read_bytes, write_bytes
98+
- fusion_enabled, wave_enabled, executor, region, etc.
6299

63100
**`tasks`** — one row per task:
64101
- run_id, group, hash, name, process, tag, status
65-
- submit, start, complete, duration, realtime (ms)
66-
- cpus, memory, rss, peak_rss, read_bytes, write_bytes
102+
- submit, start, complete, duration_ms, realtime_ms
103+
- cpus, memory_bytes, rss, peak_rss, read_bytes, write_bytes
67104
- cost, executor, machine_type, cloud_zone, exit_status
68-
- derived: runtime_ms, wait_ms, staging_ms, process_short
105+
- derived: process_short, wait_ms, staging_ms
106+
107+
**`metrics`** — per-process resource stats:
108+
- run_id, group, process
109+
- cpu/mem/vmem/time/reads/writes/cpuUsage/memUsage/timeUsage × {mean,min,q1,q2,q3,max}
69110

70111
**`costs`** (optional, from AWS CUR parquet):
71112
- run_id, process, hash, cost, used_cost, unused_cost
72113

73-
**`benchmark`** view: tasks JOIN runs LEFT JOIN costs
74-
75-
### Step 3: eCharts Static HTML Report
76-
77-
Single self-contained HTML file. Jinja2 template with embedded eCharts JS. Data injected as JSON blobs.
78-
79-
Report sections (matching current Quarto report):
80-
1. **Benchmark Overview** — grouped bars: wall time, cost, CPU time per group
81-
2. **Run Overview** — summary table of all runs
82-
3. **Process Overview** — box plots: runtime, memory, CPU per process (from /metrics ResourceData)
83-
4. **Task Overview** — scatter (runtime vs memory), timeline/gantt per run
84-
5. **Cost Overview** — stacked bars per process, used vs unused
85-
86114
## Input Format
87115

88-
Same CSV, with optional `group` column:
116+
CSV with optional `group` column:
89117

90118
```csv
91119
id,workspace,group
@@ -100,7 +128,42 @@ id,workspace,group
100128
nextflow run seqeralabs/nf-aggregate --input runs.csv --generate_benchmark_report
101129

102130
# With AWS costs
103-
nextflow run seqeralabs/nf-aggregate --input runs.csv --generate_benchmark_report --benchmark_aws_cur_report aws_cur.parquet
131+
nextflow run seqeralabs/nf-aggregate --input runs.csv --generate_benchmark_report \
132+
--benchmark_aws_cur_report aws_cur.parquet
133+
```
134+
135+
## Local Testing (individual scripts)
136+
137+
```bash
138+
# Step 1: Clean JSON
139+
uv run --with duckdb --with typer --with pyyaml python bin/clean_json.py \
140+
--data-dir modules/local/benchmark_report/tests/data \
141+
--output-dir /tmp/cleaned
142+
143+
# Step 3: Build tables
144+
uv run --with duckdb --with typer python bin/build_tables.py \
145+
--runs-csv /tmp/cleaned/runs.csv \
146+
--tasks-csv /tmp/cleaned/tasks.csv \
147+
--metrics-csv /tmp/cleaned/metrics.csv \
148+
--output-dir /tmp/tables
149+
150+
# Step 4: Render report
151+
uv run --with jinja2 --with typer --with pyyaml python bin/render_report.py \
152+
--tables-dir /tmp/tables \
153+
--brand assets/brand.yml \
154+
--output /tmp/benchmark_report.html
155+
```
156+
157+
## Running Tests
158+
159+
```bash
160+
# All new decomposed tests
161+
uv run --with duckdb --with jinja2 --with typer --with pyyaml --with pyarrow --with pytest \
162+
pytest bin/test_clean_json.py bin/test_clean_cur.py bin/test_build_tables.py bin/test_render_report.py -v
163+
164+
# Legacy monolithic tests (still work)
165+
uv run --with duckdb --with jinja2 --with typer --with pyyaml --with pyarrow --with pytest \
166+
pytest bin/test_benchmark_report.py -v
104167
```
105168

106169
## Project Structure
@@ -109,23 +172,58 @@ nextflow run seqeralabs/nf-aggregate --input runs.csv --generate_benchmark_repor
109172
nf-agg/
110173
├── workflows/nf_aggregate/main.nf # orchestrator
111174
├── modules/local/
112-
│ ├── seqera_runs_dump/ # tower-cli runs dump + metadata
113-
│ ├── benchmark_report/ # Python + DuckDB + eCharts
114-
│ └── plot_run_gantt/ # fusion-only gantt
175+
│ ├── clean_json/main.nf # JSON → CSVs
176+
│ ├── clean_cur/main.nf # CUR parquet → costs CSV
177+
│ ├── build_tables/main.nf # CSVs → query result JSONs
178+
│ ├── render_report/main.nf # JSONs → HTML report
179+
│ ├── benchmark_report/ # (legacy monolithic — deprecated)
180+
│ ├── seqera_runs_dump/ # tower-cli runs dump + metadata
181+
│ └── plot_run_gantt/ # fusion-only gantt
115182
├── lib/
116-
│ └── SeqeraApi.groovy # nf-boost request() wrappers
183+
│ └── SeqeraApi.groovy # API client (head job)
117184
├── bin/
118-
│ ├── benchmark_report.py # DuckDB + eCharts report generator
119-
│ └── plot_run_gantt.py # existing gantt plotter
120-
├── templates/
121-
│ └── benchmark_report.html # Jinja2 + eCharts template
122-
└── nextflow.config # add nf-boost plugin
185+
│ ├── clean_json.py # Step 1: normalize JSON
186+
│ ├── clean_cur.py # Step 2: normalize CUR
187+
│ ├── build_tables.py # Step 3: DuckDB queries
188+
│ ├── render_report.py # Step 4: HTML rendering
189+
│ ├── benchmark_report.py # (legacy monolithic — deprecated)
190+
│ ├── test_clean_json.py # tests for step 1
191+
│ ├── test_clean_cur.py # tests for step 2
192+
│ ├── test_build_tables.py # tests for step 3
193+
│ ├── test_render_report.py # tests for step 4
194+
│ └── test_benchmark_report.py # legacy tests
195+
└── nextflow.config
123196
```
124197

198+
## Key Params
199+
200+
| Param | Default | Purpose |
201+
|---|---|---|
202+
| `generate_benchmark_report` | false | Enable benchmark report |
203+
| `benchmark_aws_cur_report` | null | AWS CUR parquet for cost analysis |
204+
| `seqera_api_endpoint` | `https://api.cloud.seqera.io` | Platform API URL |
205+
| `skip_run_gantt` | false | Skip Gantt chart generation |
206+
| `skip_multiqc` | false | Skip MultiQC aggregation |
207+
208+
## Plugins
209+
210+
- `nf-schema@2.3.0` — param validation, samplesheet parsing
211+
- `nf-boost@0.6.0` — `request()`, `fromJson`/`toJson` for API calls
212+
213+
## Env Requirements
214+
215+
- `TOWER_ACCESS_TOKEN` — Seqera Platform API token
216+
125217
## Dependencies
126218

127219
**Nextflow plugins**: nf-boost (for `request`, `fromJson`, `toJson`)
128-
**Python container**: duckdb, jinja2, pyarrow (for parquet)
220+
**Python**: duckdb, jinja2, typer, pyarrow (for parquet), pyyaml
129221
**Not required**: R, Quarto, renv
130222

223+
## Gotchas
131224

225+
- Wave freeze strategy: `['conda', 'container', 'dockerfile']` — no `spack`
226+
- DuckDB `read_json_auto` needs file paths, not JSON strings — use temp files
227+
- `commit.gpgsign` must be true (SSH signing via 1Password)
228+
- CUR hash join: task hash contains '/' (e.g. `45/d87388`) — strip before comparing
229+
- CLEAN_CUR auto-detects CUR format (MAP vs flattened) — no user config needed

‎workflows/nf_aggregate/main.nf‎

Lines changed: 33 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,10 @@
22
// WORKFLOW: Run main seqeralabs/nf-aggregate workflow
33
//
44

5-
include { BENCHMARK_REPORT } from '../../modules/local/benchmark_report'
5+
include { CLEAN_JSON } from '../../modules/local/clean_json'
6+
include { CLEAN_CUR } from '../../modules/local/clean_cur'
7+
include { BUILD_TABLES } from '../../modules/local/build_tables'
8+
include { RENDER_REPORT } from '../../modules/local/render_report'
69
include { PLOT_RUN_GANTT } from '../../modules/local/plot_run_gantt'
710
include { SEQERA_RUNS_DUMP } from '../../modules/local/seqera_runs_dump'
811
include { MULTIQC } from '../../modules/nf-core/multiqc'
@@ -55,19 +58,17 @@ workflow NF_AGGREGATE {
5558
ch_versions = ch_versions.mix(PLOT_RUN_GANTT.out.versions)
5659

5760
//
58-
// MODULE: Generate benchmark report (v2 — API + DuckDB + eCharts)
61+
// BENCHMARK REPORT PIPELINE: API → Clean JSON → Build Tables → Render HTML
5962
//
6063
if (params.generate_benchmark_report) {
61-
aws_cur_report = params.benchmark_aws_cur_report ? Channel.fromPath(params.benchmark_aws_cur_report) : []
6264

63-
// Fetch run data directly from Seqera API using nf-boost
65+
// Step 0: Fetch run data on the head job (no process, just streaming)
6466
ch_run_data = ids.map { meta ->
6567
def data = SeqeraApi.fetchRunData(meta, seqera_api_endpoint)
6668
data.meta = [id: meta.id, workspace: meta.workspace, group: meta.group ?: 'default']
6769
return data
6870
}
6971

70-
// Write each run's data to a JSON file, collect into one directory
7172
ch_run_jsons = ch_run_data.map { data ->
7273
def json_file = file("${workDir}/run_data/${data.meta.id}.json")
7374
json_file.parent.mkdirs()
@@ -84,13 +85,36 @@ workflow NF_AGGREGATE {
8485
return dir
8586
}
8687

87-
BENCHMARK_REPORT(
88-
ch_data_dir,
89-
aws_cur_report,
88+
// Step 1: Clean raw JSON → normalized CSVs (runs, tasks, metrics)
89+
CLEAN_JSON(ch_data_dir)
90+
ch_versions = ch_versions.mix(CLEAN_JSON.out.versions)
91+
92+
// Step 2: Clean AWS CUR parquet → costs CSV (optional)
93+
if (params.benchmark_aws_cur_report) {
94+
ch_cur = Channel.fromPath(params.benchmark_aws_cur_report)
95+
CLEAN_CUR(ch_cur)
96+
ch_costs_csv = CLEAN_CUR.out.costs_csv
97+
ch_versions = ch_versions.mix(CLEAN_CUR.out.versions)
98+
} else {
99+
ch_costs_csv = Channel.fromPath("${projectDir}/assets/NO_FILE", checkIfExists: false).ifEmpty([])
100+
}
101+
102+
// Step 3: Build query result tables from CSVs
103+
BUILD_TABLES(
104+
CLEAN_JSON.out.runs_csv,
105+
CLEAN_JSON.out.tasks_csv,
106+
CLEAN_JSON.out.metrics_csv.ifEmpty(file("${projectDir}/assets/NO_FILE")),
107+
ch_costs_csv.ifEmpty(file("${projectDir}/assets/NO_FILE")),
108+
)
109+
ch_versions = ch_versions.mix(BUILD_TABLES.out.versions)
110+
111+
// Step 4: Render HTML report from pre-computed tables
112+
RENDER_REPORT(
113+
BUILD_TABLES.out.tables_dir,
90114
file("${projectDir}/assets/brand.yml", checkIfExists: true),
91115
file("${projectDir}/assets/seqera_logo_color.svg", checkIfExists: true),
92116
)
93-
ch_versions = ch_versions.mix(BENCHMARK_REPORT.out.versions)
117+
ch_versions = ch_versions.mix(RENDER_REPORT.out.versions)
94118
}
95119

96120
//

0 commit comments

Comments
 (0)