Merge pull request #24 from redhat-performance/feature/fix-fio-field-limit-remove-per-job

Fix FIO field count exceeding OpenSearch 5,000 field limit

Author: grdumas

.gitignore

@@ -38,4 +38,14 @@ temp_*/
 
 # Config files (credentials)
 config/export_config.yml
-src/chronicler/config/export_config.yml
+src/chronicler/config/export_config.yml
+
+# Temporary analysis files (issue #19)
+FIELD_COUNT_ANALYSIS.md
+IMPLEMENTATION_PLAN_RPOPC-1273.md
+analyze_run_fields.py
+count_fields_by_section.py
+show_metrics_structure.py
+verify_field_count.py
+backups/
+sample_data/

README.md

@@ -230,7 +230,7 @@ podman run --rm \
 |-----------|------------------------|-----------|-------|
 | CoreMark | Supported | `coremark_processor.py` | Single-thread CPU performance |
 | CoreMark Pro | Supported | `coremark_pro_processor.py` | 9 workload types |
-| FIO | Supported | `fio_processor.py` | Flexible I/O tester |
+| FIO | Supported | `fio_processor.py` | Flexible I/O tester (see [per-job data docs](docs/fio-per-job-data.md)) |
 | HPL (autohpl) | Supported | `autohpl_processor.py` | High Performance Computing Linpack |
 | Passmark | Supported | `passmark_processor.py` | CPU & Memory marks |
 | Phoronix Test Suite | Supported | `phoronix_processor.py` | 51 sub-tests (BOPs) |

docs/fio-per-job-data.md

@@ -0,0 +1,171 @@
+# FIO Per-Job Data: OpenSearch vs Raw Archives
+
+## Overview
+
+FIO benchmark results contain both **aggregated metrics** (totals across all jobs/disks) and **per-job breakdown** (individual disk performance). Due to OpenSearch's 5,000 field limit, only aggregated data is exported to the `zathras-results` index. Per-job data remains available in the raw JSON archives.
+
+## What's in OpenSearch (zathras-results)
+
+Each FIO run in OpenSearch contains:
+
+### Aggregated Metrics
+- **Bandwidth**: total_bandwidth_kbps (min/max/mean)
+- **IOPS**: total_iops (min/max/mean)
+- **Latency**: avg_latency_mean_ns, avg_clat_mean_ns, avg_slat_mean_ns (with min/max/stddev)
+- **Latency percentiles**: p1, p5, p10, p50, p90, p95, p99, p99.5, p99.9
+- **I/O totals**: total_io_bytes, total_ios
+- **CPU**: avg_cpu_usr_pct, avg_cpu_sys_pct
+- **Metadata**: num_jobs, num_disks
+
+### Timeseries Summary
+- Statistical summary of timeseries data: count, mean, min, max, stddev
+
+### Configuration
+- All FIO test parameters and settings
+
+**Use OpenSearch for**: Aggregate performance trends, run comparisons, dashboard visualizations
+
+## What's in Raw JSON Archives
+
+The full JSON documents (not exported to OpenSearch) additionally contain:
+
+### Per-Job Details (`metrics.jobs` array)
+
+For **each disk/job**:
+- Job metadata: job_number, jobname, device path, elapsed_seconds
+- Read metrics (if read test):
+  - Bandwidth: kbps, min, max, mean, stddev, aggregate %
+  - IOPS: value, min, max, mean, stddev
+  - Latency: mean, min, max, stddev (regular + clat + slat)
+  - **Latency percentiles**: p1, p5, p10, p50, p90, p95, p99, p99.5, p99.9
+  - I/O: bytes, count, runtime
+  - CPU: usr%, sys%
+  - **I/O depth distribution**: % at depth 1, 2, 4, 8, 16, 32, 64+
+  - **Latency distribution buckets**: microsecond and millisecond ranges
+- Write metrics (if write test): same structure
+- Mixed metrics (if mixed test): both read and write
+
+### Full Timeseries Data
+- Every timeseries point with timestamp and metrics
+- Available in separate `zathras-timeseries` index (if enabled)
+
+**Use raw JSON for**: Per-disk analysis, identifying slow disks, latency distribution analysis
+
+## Accessing Per-Job Data
+
+### Method 1: Direct File Read
+
+Raw JSON documents are stored in the same location as the benchmark archives:
+
+```python
+import json
+from pathlib import Path
+
+# Load the full document
+json_path = Path("/path/to/archive/fio-results.json")
+with open(json_path) as f:
+    doc = json.load(f)
+
+# Access per-job data for a specific run
+jobs = doc["results"]["runs"]["run_0"]["metrics"]["jobs"]
+
+for job in jobs:
+    print(f"Device: {job['device']}")
+    print(f"  Bandwidth: {job['read']['bandwidth_kbps']} kbps")
+    print(f"  IOPS: {job['read']['iops']}")
+    print(f"  P99 Latency: {job['read']['latency_percentiles']['p99']} ns")
+```
+
+### Method 2: Programmatic Access (Python API)
+
+```python
+from chronicler.processors.fio_processor import FioProcessor
+
+# Process with full detail
+processor = FioProcessor("/path/to/benchmark/archive")
+document = processor.process()
+
+# Get full dict (includes per-job data)
+full_dict = document.to_dict()
+
+# Access per-job data
+jobs = full_dict["results"]["runs"]["run_0"]["metrics"]["jobs"]
+```
+
+### Method 3: Query Pattern for Analysis
+
+Example script to find slow disks across multiple test runs:
+
+```python
+import json
+from pathlib import Path
+
+def find_slow_disks(json_path, p99_threshold_ns=500_000):
+    """Find disks with p99 latency above threshold."""
+    with open(json_path) as f:
+        doc = json.load(f)
+    
+    slow_disks = []
+    for run_key, run_data in doc["results"]["runs"].items():
+        if "metrics" not in run_data or "jobs" not in run_data["metrics"]:
+            continue
+        
+        for job in run_data["metrics"]["jobs"]:
+            device = job.get("device", "unknown")
+            read_data = job.get("read", {})
+            p99 = read_data.get("latency_percentiles", {}).get("p99")
+            
+            if p99 and p99 > p99_threshold_ns:
+                slow_disks.append({
+                    "run": run_key,
+                    "device": device,
+                    "p99_latency_ns": p99,
+                    "bandwidth_kbps": read_data.get("bandwidth_kbps"),
+                    "iops": read_data.get("iops"),
+                })
+    
+    return slow_disks
+
+# Usage
+slow = find_slow_disks("fio-results.json", p99_threshold_ns=500_000)
+for disk in slow:
+    print(f"{disk['device']} in {disk['run']}: p99={disk['p99_latency_ns']}ns")
+```
+
+## Why Not Store Per-Job Data in OpenSearch?
+
+### Design Decision
+
+FIO was the only benchmark that stored per-instance breakdown in OpenSearch. Other benchmarks (CoreMark, Passmark, Uperf) follow an aggregated approach:
+- **CoreMark**: Aggregate across threads, not per-thread
+- **Passmark**: Aggregate across iterations, not per-iteration  
+- **Uperf**: Aggregate across workers, not per-worker
+
+To maintain consistency and stay within OpenSearch's 5,000 field limit, FIO now follows the same pattern.
+
+### Field Count Impact
+
+**With per-job data** (48 runs, 1 job each):
+- Fields: ~6,632
+- Status: ❌ Exceeds 5,000 limit
+
+**Without per-job data**:
+- Fields: ~3,176  
+- Status: ✅ Under 5,000 limit (36% headroom)
+
+### Future: Separate Per-Job Index
+
+If per-job querying in OpenSearch becomes a frequent need, a separate `zathras-fio-job-timeseries` index could be implemented (similar to how general timeseries data is handled). See [GitHub issue #19](https://github.com/redhat-performance/chronicler/issues/19) for discussion.
+
+## Summary
+
+| Data Type | OpenSearch | Raw JSON |
+|-----------|------------|----------|
+| Aggregated metrics | ✅ | ✅ |
+| Timeseries summary | ✅ | ✅ |
+| Configuration | ✅ | ✅ |
+| Per-job breakdown | ❌ | ✅ |
+| Full timeseries | ❌ | ✅ |
+
+**For most analysis**: Use OpenSearch (fast queries, dashboards)  
+**For per-disk troubleshooting**: Use raw JSON archives (full granularity)

src/chronicler/processors/README.md

@@ -29,7 +29,8 @@ BaseProcessor (abstract)
 ├── UperfProcessor
 ├── PigProcessor
 ├── AutoHPLProcessor
-└── SpecCPU2017Processor
+├── SpecCPU2017Processor
+└── FioProcessor
 ```
 
 ### Data Flow
@@ -417,6 +418,56 @@ Results:
 
 ---
 
+### 12. FIO (`fio_processor.py`)
+
+**Benchmarks:** Flexible I/O Tester - disk performance
+
+**Key Features:**
+- Parses multiple workload runs (different I/O patterns)
+- Extracts bandwidth, IOPS, latency metrics aggregated across all jobs
+- Stores latency percentiles (p1, p5, p10, p50, p90, p95, p99, p99.5, p99.9)
+- **Per-job breakdown removed from OpenSearch** (available in raw JSON)
+
+**Data Structure:**
+```python
+Run:
+  metrics:
+    # Aggregated across all jobs/disks
+    total_bandwidth_kbps: 1000000
+    total_iops: 250000
+    avg_latency_mean_ns: 134845
+    avg_clat_mean_ns: 131462
+    avg_slat_mean_ns: 3382
+    # Latency percentiles (aggregate)
+    avg_latency_p1_ns: 72192
+    avg_latency_p50_ns: 128512
+    avg_latency_p99_ns: 259072
+    # Metadata
+    num_jobs: 8
+    num_disks: 8
+  timeseries_summary:
+    count: 120
+    mean: 473231.0
+    min: 465628.0
+    max: 490332.0
+  configuration:
+    operation: "read"
+    block_size: "4k"
+    iodepth: 16
+```
+
+**Field Count:** ~3,200 fields for 48 runs (well under 5,000 limit)
+
+**Design Decision:**
+- FIO originally stored per-job breakdown (`metrics.jobs` array) in OpenSearch
+- This caused field explosion (6,632 fields for 48 runs with per-job data)
+- Changed to aggregated-only approach (matching CoreMark, Passmark, Uperf)
+- Per-job data preserved in raw JSON archives
+
+See `docs/fio-per-job-data.md` for accessing per-job breakdowns from raw archives.
+
+---
+
 ## Data Organization
 
 ### Run Structure

src/chronicler/schema.py

@@ -406,18 +406,30 @@ def calculate_content_hash(self, exclude_processing_timestamp: bool = True) -> s
 
     def to_dict_summary_only(self) -> Dict[str, Any]:
         """
-        Convert to dictionary WITHOUT timeseries data.
-        Only includes timeseries_summary for each run.
+        Convert to dictionary WITHOUT timeseries data and per-job details.
+        Only includes timeseries_summary and aggregated metrics for each run.
         Used for the main zathras-results index.
+
+        Removes:
+        - timeseries: Detailed time series data (available in zathras-timeseries index)
+        - metrics.jobs: Per-job breakdown (available in raw JSON archives)
+
+        This keeps field count under OpenSearch's default 5,000 field limit and
+        aligns FIO with the aggregated approach used by other benchmarks.
         """
         result = self.to_dict()
 
-        # Remove timeseries from all runs
+        # Remove timeseries and per-job details from all runs
         if 'results' in result and 'runs' in result['results']:
-            for run_key, run_data in result['results']['runs'].items():
+            for _, run_data in result['results']['runs'].items():
+                # Remove timeseries data
                 if 'timeseries' in run_data:
                     del run_data['timeseries']
 
+                # Remove per-job breakdown (FIO-specific)
+                if 'metrics' in run_data:
+                    run_data['metrics'].pop('jobs', None)
+
         return result
 
     def extract_timeseries_documents(self) -> List['TimeSeriesDocument']:

tests/test_schema.py

@@ -313,6 +313,55 @@ def test_to_dict_summary_only_removes_timeseries(self, full_document):
         run_data = d["results"]["runs"]["run_1"]
         assert "timeseries" not in run_data
 
+    def test_to_dict_summary_only_removes_per_job_details(self):
+        """Test that to_dict_summary_only removes metrics.jobs array (FIO-specific)."""
+        doc = ZathrasDocument(
+            metadata=Metadata(document_id="fio-test"),
+            test=TestInfo(name="fio", version="3.35"),
+            system_under_test=SystemUnderTest(),
+            test_configuration=TestConfiguration(),
+            results=Results(
+                status="PASS",
+                runs={
+                    "run_0": Run(
+                        run_number=0,
+                        status="PASS",
+                        metrics={
+                            "total_bandwidth_kbps": 1000000,
+                            "total_iops": 250000,
+                            "jobs": [
+                                {
+                                    "job_number": 0,
+                                    "device": "/dev/sda",
+                                    "bandwidth_kbps": 500000,
+                                    "iops": 125000,
+                                },
+                                {
+                                    "job_number": 1,
+                                    "device": "/dev/sdb",
+                                    "bandwidth_kbps": 500000,
+                                    "iops": 125000,
+                                },
+                            ],
+                        },
+                    )
+                },
+            ),
+        )
+
+        # Full dict should have jobs
+        full_dict = doc.to_dict()
+        assert "jobs" in full_dict["results"]["runs"]["run_0"]["metrics"]
+        assert len(full_dict["results"]["runs"]["run_0"]["metrics"]["jobs"]) == 2
+
+        # Summary dict should NOT have jobs
+        summary_dict = doc.to_dict_summary_only()
+        assert "jobs" not in summary_dict["results"]["runs"]["run_0"]["metrics"]
+
+        # But should still have aggregated metrics
+        assert summary_dict["results"]["runs"]["run_0"]["metrics"]["total_bandwidth_kbps"] == 1000000
+        assert summary_dict["results"]["runs"]["run_0"]["metrics"]["total_iops"] == 250000
+
     def test_validate_valid_document(self, minimal_document):
         is_valid, errors = minimal_document.validate()
         assert is_valid