refactor(records): remove legacy results processors; route by record type#1144
Open
ajcasagrande wants to merge 1 commit into
Open
refactor(records): remove legacy results processors; route by record type#1144ajcasagrande wants to merge 1 commit into
ajcasagrande wants to merge 1 commit into
Conversation
…type Remove the legacy `results_processor` plugin category entirely and route records dynamically by `record_type` through accumulators and stream exporters. Taxonomy - Delete the `results_processor` category and its registrations. Behavior moves to `accumulator` / `stream_exporter` / `analyzer` / `data_exporter`. - Delete legacy `metric_results_processor`, `timeslice_metric_results_processor`, `record_export_results_processor`, and the bespoke `network_latency/protocols`. - Rename `otel_metrics_results_processor` -> `otel_metrics_streamer` and `accuracy_results_processor` -> the accuracy accumulator. RecordsManager - Replace the per-type processor lists and fan-out methods with a routing table built from plugin metadata and a single `_dispatch_record(record)` keyed on `record.record_type`. DERIVED metrics now resolve in MetricsAccumulator. Accuracy - Move per-record grade transport off the `accuracy.` summary namespace to underscore keys (`accuracy_correct` / `accuracy_unparsed`) so the registered transport metrics no longer collide with the `accuracy.` summary tags (which had been dropping the overall unparsed count). Network-adjusted metrics - Wire `network_adjusted_*` latency metrics into per-timeslice results (they were only in the overall summary). The per-record adjustment `max(latency - rtt, 0)` is window-independent, so it is computed once over each full source column and both the overall summary and every timeslice aggregate masked views -- no redundant per-window subtraction, and no double-application (the analyzer only reads the column store, never writes back). Full unit suite: 14422 passed, 94 skipped, 1 xfailed. Plugin artifacts + schemas validate; every registered plugin class imports. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
ajcasagrande
requested review from
FrankD412,
debermudez,
ganeshku1,
ilana-n,
lkomali and
matthewkotila
as code owners
Try out this PRQuick install: pip install --upgrade --force-reinstall git+https://github.com/ai-dynamo/aiperf.git@17dabb5c97da8eba6afb113e2d4a02b06e552c00Recommended with virtual environment (using uv): uv venv --python 3.12 && source .venv/bin/activate
uv pip install --upgrade --force-reinstall git+https://github.com/ai-dynamo/aiperf.git@17dabb5c97da8eba6afb113e2d4a02b06e552c00Last updated for commit: |
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
Contributor
WalkthroughThis PR replaces the legacy per-category "results processor" architecture with an accumulator/stream-exporter model. Records now carry a Estimated code review effort: 5 (Critical) | ~120 minutes ChangesAccumulator/Router Migration
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tests/unit/post_processors/test_metrics_accumulator.py (1)
27-28: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAnnotate
mock_runin this test module.
mock_runis aBenchmarkRunfixture, so the test methods should type it asBenchmarkRuninstead of leaving it implicit.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/post_processors/test_metrics_accumulator.py` around lines 27 - 28, Annotate the test methods’ mock_run fixture parameter as BenchmarkRun, importing BenchmarkRun if needed; update the relevant tests in the metrics accumulator module without changing fixture behavior.Source: Coding guidelines
🧹 Nitpick comments (3)
tests/unit/common/models/test_record_models.py (1)
20-21: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConform the updated tests to the repository test convention.
Add
-> Noneand include the expected outcome in each test name.Proposed naming and annotation update
- def test_profile_results_with_timeslices(self): + def test_profile_results_timeslices_preserves_metric_lookup(self) -> None: - def test_profile_results_without_timeslices(self): + def test_profile_results_no_timeslices_defaults_to_none(self) -> None: - def test_profile_results_with_empty_timeslices(self): + def test_profile_results_empty_timeslices_preserves_empty_list(self) -> None: - def test_profile_results_with_multiple_timeslices_and_metrics(self): + def test_profile_results_multiple_timeslices_maps_metrics_by_tag(self) -> None:As per coding guidelines, “Add type hints to all functions” and “Name tests as
test_<function>_<scenario>_<expected>.”Also applies to: 61-62, 80-81, 100-101
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/common/models/test_record_models.py` around lines 20 - 21, Update the affected test methods, including test_profile_results_with_timeslices and the tests at the other referenced locations, to return-annotate with -> None and rename each using test_<function>_<scenario>_<expected>, explicitly stating the expected outcome.Source: Coding guidelines
tests/unit/post_processors/test_timeslice_metric_results_processor.py (1)
21-23: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAnnotate the
mock_runfixture parameter.The newly added tests leave
mock_rununtyped. Add its concrete fixture type consistently across these methods. As per coding guidelines, “Add type hints to all functions, including parameters and return values.”Also applies to: 29-33, 35-36, 64-65, 87-88, 113-114, 141-142, 172-175, 183-186, 209-210
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/post_processors/test_timeslice_metric_results_processor.py` around lines 21 - 23, Annotate the mock_run parameter in all affected test methods, including test_initialization_without_slice_duration_disables_timeslices, with the concrete fixture type used by the mock_run fixture; keep the annotations consistent across every listed test and preserve the existing return type hints.Source: Coding guidelines
src/aiperf/accuracy/models.py (1)
11-25: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReduce redundant constant commentary.
Most of this block restates the constants and processing flow. Retain the collision rationale, but condense the “what” details.
As per coding guidelines, “Write comments only to explain ‘why’, not ‘what’; do not over-comment code.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aiperf/accuracy/models.py` around lines 11 - 25, Condense the comments above the accuracy constants to explain only the namespace-collision rationale. Remove descriptions of emitted tags, processing flow, and record contents, while retaining that transport keys use the accuracy_ prefix to avoid colliding with accuracy. summary tags, particularly accuracy.unparsed.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/plugins/plugin-system.md`:
- Around line 137-138: Update the plugin-category total in the introduction of
plugin-system.md from 33 to 32 so it matches the table entries after replacing
accumulator and stream_exporter.
In `@src/aiperf/metrics/types/network_adjusted_metrics.py`:
- Around line 11-14: The documentation around MetricsAccumulator and
_derive_value incorrectly states percentile, mean, and standard-deviation
invariants unconditionally. Clarify that these properties apply only when
shifted values remain nonnegative; note that adjusted values are clamped at zero
when RTT exceeds source latency, so clamping can alter the distributions.
In `@src/aiperf/records/records_manager.py`:
- Around line 535-549: Capture the errors returned by
_dispatch_record(record_data) and propagate them through the phase-level error
handling before calling _records_tracker.update_from_record_data(record_data).
Ensure metric-handler failures are included in
_error_tracker.increment_error_count_for_phase (or cause phase failure) so
records are not marked successfully processed when dispatching fails.
In `@tests/unit/gpu_telemetry/test_telemetry_integration.py`:
- Around line 272-277: Annotate the failing_process_telemetry_record replacement
coroutine with the same record parameter type and async return type as the
original process_telemetry_record method, preserving the existing behavior while
satisfying type checking.
---
Outside diff comments:
In `@tests/unit/post_processors/test_metrics_accumulator.py`:
- Around line 27-28: Annotate the test methods’ mock_run fixture parameter as
BenchmarkRun, importing BenchmarkRun if needed; update the relevant tests in the
metrics accumulator module without changing fixture behavior.
---
Nitpick comments:
In `@src/aiperf/accuracy/models.py`:
- Around line 11-25: Condense the comments above the accuracy constants to
explain only the namespace-collision rationale. Remove descriptions of emitted
tags, processing flow, and record contents, while retaining that transport keys
use the accuracy_ prefix to avoid colliding with accuracy. summary tags,
particularly accuracy.unparsed.
In `@tests/unit/common/models/test_record_models.py`:
- Around line 20-21: Update the affected test methods, including
test_profile_results_with_timeslices and the tests at the other referenced
locations, to return-annotate with -> None and rename each using
test_<function>_<scenario>_<expected>, explicitly stating the expected outcome.
In `@tests/unit/post_processors/test_timeslice_metric_results_processor.py`:
- Around line 21-23: Annotate the mock_run parameter in all affected test
methods, including
test_initialization_without_slice_duration_disables_timeslices, with the
concrete fixture type used by the mock_run fixture; keep the annotations
consistent across every listed test and preserve the existing return type hints.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: adf06611-66a8-45da-921b-35ac6e9ce52a
📒 Files selected for processing (68)
docs/accuracy/accuracy_stubs.mddocs/cli-options.mddocs/dev/patterns.mddocs/diagrams/metrics-flow.mddocs/plugins/plugin-system.mddocs/reference/list-metric-aggregation.mdsrc/aiperf/accuracy/accuracy_record_processor.pysrc/aiperf/accuracy/accuracy_results_processor.pysrc/aiperf/accuracy/models.pysrc/aiperf/common/messages/inference_messages.pysrc/aiperf/common/models/credit_models.pysrc/aiperf/common/models/network_latency_models.pysrc/aiperf/common/models/record_models.pysrc/aiperf/common/models/server_metrics_models.pysrc/aiperf/common/models/telemetry_models.pysrc/aiperf/config/artifacts.pysrc/aiperf/exporters/metrics_base_exporter.pysrc/aiperf/gpu_telemetry/protocols.pysrc/aiperf/metrics/accumulator.pysrc/aiperf/metrics/base_aggregate_metric.pysrc/aiperf/metrics/derived_sum_metric.pysrc/aiperf/metrics/display_units.pysrc/aiperf/metrics/network_adjusted_analyzer.pysrc/aiperf/metrics/types/accuracy_metrics.pysrc/aiperf/metrics/types/network_adjusted_metrics.pysrc/aiperf/metrics/types/power_efficiency_metrics.pysrc/aiperf/network_latency/accumulator.pysrc/aiperf/network_latency/jsonl_writer.pysrc/aiperf/network_latency/protocols.pysrc/aiperf/orchestrator/search_planner/_pooled_percentile.pysrc/aiperf/plugin/categories.yamlsrc/aiperf/plugin/enums.pysrc/aiperf/plugin/plugins.pysrc/aiperf/plugin/plugins.yamlsrc/aiperf/plugin/schema/plugins.schema.jsonsrc/aiperf/post_processors/metric_results_processor.pysrc/aiperf/post_processors/otel_metrics_results_processor.pysrc/aiperf/post_processors/protocols.pysrc/aiperf/post_processors/record_export_jsonl_writer.pysrc/aiperf/post_processors/record_export_results_processor.pysrc/aiperf/post_processors/strategies/core.pysrc/aiperf/post_processors/timeslice_metric_results_processor.pysrc/aiperf/records/records_manager.pysrc/aiperf/records/records_manager_processing.pysrc/aiperf/server_metrics/protocols.pytests/unit/accuracy/test_accuracy_record_processor.pytests/unit/common/models/test_record_models.pytests/unit/exporters/conftest.pytests/unit/gpu_telemetry/test_telemetry_integration.pytests/unit/metrics/test_list_metric_aggregation.pytests/unit/metrics/test_power_efficiency_metrics.pytests/unit/network_latency/test_jsonl_writer.pytests/unit/post_processors/conftest.pytests/unit/post_processors/test_metric_results_processor.pytests/unit/post_processors/test_metrics_accumulator.pytests/unit/post_processors/test_network_adjusted_metrics.pytests/unit/post_processors/test_otel_error_paths.pytests/unit/post_processors/test_otel_metrics_results_processor.pytests/unit/post_processors/test_post_processor_integration.pytests/unit/post_processors/test_record_export_jsonl_writer.pytests/unit/post_processors/test_record_export_jsonl_writer_detailed.pytests/unit/post_processors/test_timeslice_metric_results_processor.pytests/unit/property/_numeric_bounds_baseline.txttests/unit/records/test_records_manager.pytests/unit/records/test_records_manager_network_latency.pytests/unit/records/test_records_manager_process_results.pytests/unit/records/test_records_manager_routing.pytests/unit/server_metrics/test_accumulator.py
💤 Files with no reviewable changes (10)
- src/aiperf/post_processors/metric_results_processor.py
- src/aiperf/network_latency/protocols.py
- src/aiperf/post_processors/timeslice_metric_results_processor.py
- src/aiperf/post_processors/record_export_results_processor.py
- tests/unit/exporters/conftest.py
- tests/unit/property/_numeric_bounds_baseline.txt
- src/aiperf/post_processors/record_export_jsonl_writer.py
- src/aiperf/plugin/categories.yaml
- src/aiperf/records/records_manager_processing.py
- src/aiperf/plugin/schema/plugins.schema.json
Comment on lines
+137
to
+138
| | `accumulator` | `AccumulatorType` | Record-type-routed aggregation and summary computation | | ||
| | `stream_exporter` | `StreamExporterType` | Record-type-routed streaming sinks such as JSONL and OpenTelemetry | |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the plugin-category total.
These replacements leave 32 table entries, but the introduction still says 33. Update that count to prevent the reference from contradicting itself.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/plugins/plugin-system.md` around lines 137 - 138, Update the
plugin-category total in the introduction of plugin-system.md from 33 to 32 so
it matches the table entries after replacing accumulator and stream_exporter.
Comment on lines
+11
to
+14
| known at the end of the run, so MetricsAccumulator injects the shifted distributions | ||
| after aggregation (subtracting a constant shifts every percentile and the mean by that | ||
| constant and leaves the standard deviation unchanged). ``_derive_value`` therefore | ||
| always defers. |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Document the clamping caveat.
The stated percentile/mean/std invariants are false when RTT exceeds a source latency: adjusted values are clamped at zero. Limit this claim to unclamped distributions.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/aiperf/metrics/types/network_adjusted_metrics.py` around lines 11 - 14,
The documentation around MetricsAccumulator and _derive_value incorrectly states
percentile, mean, and standard-deviation invariants unconditionally. Clarify
that these properties apply only when shifted values remain nonnegative; note
that adjusted values are clamped at zero when RTT exceeds source latency, so
clamping can alter the distributions.
Comment on lines
+535
to
+549
| await self._dispatch_record(record_data) | ||
| self._records_tracker.update_from_record_data(record_data) | ||
| if record_data.error: | ||
| self._error_tracker.increment_error_count_for_phase( | ||
| record_data.metadata.benchmark_phase, record_data.error | ||
| ) | ||
|
|
||
| Per-handler exceptions are caught so one bad accumulator does not abort | ||
| the others. GPU telemetry / server metrics records are routed via their | ||
| own ``@on_pull_message`` handlers (and main's side-channel processors) | ||
| and do not flow through here. | ||
| """ | ||
| targets: list[object] = [ | ||
| *self._metric_record_accumulators, | ||
| *self._metric_record_stream_exporters, | ||
| ] | ||
| if not targets: | ||
| return | ||
| results = await asyncio.gather( | ||
| *[t.process_record(record_data) for t in targets], | ||
| return_exceptions=True, | ||
| ) | ||
| for target, result in zip(targets, results, strict=True): | ||
| if isinstance(result, BaseException): | ||
| self.error( | ||
| f"Accumulator {target.__class__.__name__} failed for " | ||
| f"metric_records: {result!r}" | ||
| ) | ||
| phase = record_data.metadata.benchmark_phase | ||
| if ( | ||
| phase in self._complete_credit_phases | ||
| and self._records_tracker.check_and_set_all_records_received_for_phase( | ||
| phase | ||
| ) | ||
| ): | ||
| await self._handle_all_records_received(phase) |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Preserve metric-handler failures before counting the record.
_dispatch_record() returns accumulator/exporter exceptions, but this path discards them and immediately marks the record processed. A failed metric accumulator can therefore yield incomplete metrics while completed advances and the phase error summary omits the failure. Record these errors in the phase-level error path, or fail the phase before updating the tracker.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/aiperf/records/records_manager.py` around lines 535 - 549, Capture the
errors returned by _dispatch_record(record_data) and propagate them through the
phase-level error handling before calling
_records_tracker.update_from_record_data(record_data). Ensure metric-handler
failures are included in _error_tracker.increment_error_count_for_phase (or
cause phase failure) so records are not marked successfully processed when
dispatching fails.
Comment on lines
+272
to
+277
| async def failing_process_telemetry_record(record): | ||
| if record.gpu_index == 0: # Fail on first GPU only | ||
| raise ValueError("Simulated processing error") | ||
| return await original_process(record) | ||
|
|
||
| faulty_processor.process_telemetry_record = failing_process_result | ||
| faulty_processor.process_telemetry_record = failing_process_telemetry_record |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Annotate the replacement coroutine.
failing_process_telemetry_record introduces untyped parameter and return contracts.
Proposed fix
- async def failing_process_telemetry_record(record):
+ async def failing_process_telemetry_record(
+ record: TelemetryRecord,
+ ) -> None:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| async def failing_process_telemetry_record(record): | |
| if record.gpu_index == 0: # Fail on first GPU only | |
| raise ValueError("Simulated processing error") | |
| return await original_process(record) | |
| faulty_processor.process_telemetry_record = failing_process_result | |
| faulty_processor.process_telemetry_record = failing_process_telemetry_record | |
| async def failing_process_telemetry_record( | |
| record: TelemetryRecord, | |
| ) -> None: | |
| if record.gpu_index == 0: # Fail on first GPU only | |
| raise ValueError("Simulated processing error") | |
| return await original_process(record) | |
| faulty_processor.process_telemetry_record = failing_process_telemetry_record |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/unit/gpu_telemetry/test_telemetry_integration.py` around lines 272 -
277, Annotate the failing_process_telemetry_record replacement coroutine with
the same record parameter type and async return type as the original
process_telemetry_record method, preserving the existing behavior while
satisfying type checking.
Source: Coding guidelines
ilana-n
reviewed
Jul 13, 2026
ilana-n
left a comment
ilana-n
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the records pipeline cleanup! This is great + much cleaner. Going to take some more time to understand whats going on but for now one comment about seperating phase exports
Contributor
There was a problem hiding this comment.
This now phase-scopes only MetricsAccumulator; AccuracyResultsProcessor is also a metric_records accumulator in this PR, but it falls through to plain summarize(). Since AccuracyResultsProcessor.process_record() increments one global counter set for every record, warmup records are included in the profiling accuracy summary, and the warmup summary can include profiling records too. I reproduced this by feeding one correct warmup record and one incorrect profiling record: accuracy.overall came back as count=2/current=0.5 instead of the profiling-only count=1/current=0.0. Please either make accuracy implement a phase-scoped export path or store counts keyed by phase and summarize the requested phase here.
FrankD412
approved these changes
Jul 15, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Removes the legacy
results_processorplugin category entirely and routes records dynamically byrecord_typethrough accumulators and stream exporters. Net +1,634 / −3,857 across 68 files.Taxonomy
results_processorcategory and its registrations. Behavior moves toaccumulator/stream_exporter/analyzer/data_exporter.metric_results_processor,timeslice_metric_results_processor,record_export_results_processor, and the bespokenetwork_latency/protocols.otel_metrics_results_processor→otel_metrics_streamer, andaccuracy_results_processor→ the accuracy accumulator.RecordsManager
_dispatch_record(record)keyed onrecord.record_type. DERIVED metrics now resolve inMetricsAccumulator.Accuracy
accuracy.summary namespace to underscore keys (accuracy_correct/accuracy_unparsed) so the registered transport metrics no longer collide with theaccuracy.summary tags — a collision that had been dropping the overall unparsed count.Network-adjusted metrics
network_adjusted_*latency metrics into per-timeslice results (previously only in the overall summary). The per-record adjustmentmax(latency − rtt, 0)is window-independent, so it is computed once over each full source column and both the overall summary and every timeslice aggregate masked views — no redundant per-window subtraction, and no double-application (the analyzer only reads the column store, never writes back).Verification
record_typerouting coverage complete:metric_records,gpu_telemetry,server_metrics,network_latency,credit_phase_statsall have consumers.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Documentation