perf: streaming analyze — flat memory at any input size#10
Merged
Conversation
…B on 100MB file) The pipeline materialized every event twice (LogEvent list + classified list) plus two full sorts, despite the result keeping only counters, bounded action-item groups, per-device counts, and 300 events. 100MB syslog measured 651MB peak RSS; 1GB extrapolated to ~6.5GB + worker timeout death (verified review finding). - classifier.iter_classify(): lazy per-event classification (new building block); classify_events() unchanged contract on top of it - analyzer._aggregate_stream(): single bounded pass — per-severity top-K heaps keyed (timestamp, -seq) that reproduce classify_events' (severity, ts desc, arrival) ordering EXACTLY; action-item groups keep the 5 newest sample messages via the same bounded-heap scheme - analyze() accepts any Iterable[LogEvent]; build_action_items / _extract_top_devices keep their signatures (now streaming inside) - web /api/analyze file+directory sources pass generators end-to-end; peek-based empty check works for lists and generators alike; new size guard AI_LOG_ANALYZER_MAX_FILE_MB (default 200) returns 413 with CLI guidance — streaming fixes memory, not the 120s worker CPU budget - cli analyze chains lazy iterators — multi-GB files now CLI-safe Measured (100MB synthetic syslog, 1.04M events, identical outputs): before: 651MB peak RSS, 41.1s after: 40MB peak RSS, 40.8s Tests: 270 -> 278 (ordering-contract equivalence vs classify_events, generator==list results, aggregator bound invariants, 413 guard, empty-file 404, CLI happy path + exit-2 — cli.py was 0% covered)
Reviewer's GuideRefactors the analyze pipeline to be fully streaming and bounded-memory by introducing a single-pass aggregator over a lazy classifier, updating action-item and device aggregation helpers to be streaming-safe, wiring web and CLI entrypoints to pass generators instead of materialized lists with a size guard for the web route, and adding tests to pin behavioral equivalence and bounds plus documentation updates. Sequence diagram for streaming analyze pipeline over lazy classifiersequenceDiagram
actor User
participant WebApp as api_analyze
participant CLI as cmd_analyze
participant Analyzer as analyze
participant Classifier as iter_classify
participant Aggregator as _aggregate_stream
participant ActionItems as _finalize_action_items
participant TopDevices as _finalize_top_devices
User->>WebApp: HTTP POST /api/analyze
WebApp->>WebApp: parse_file / parse_directory
WebApp->>Analyzer: analyze(events_generator, use_llm)
User->>CLI: ai-log-analyzer analyze ...
CLI->>CLI: parse_file / frr_docker_logs / parse_lines
CLI->>Analyzer: analyze(events_generator, use_llm)
Analyzer->>Classifier: iter_classify(events_generator)
Classifier-->>Analyzer: ClassifiedEvent (streamed)
Analyzer->>Aggregator: _aggregate_stream(iter_classify(events))
Aggregator->>Aggregator: _group_actionable
Aggregator->>Aggregator: _count_device
Aggregator-->>Analyzer: top_events, sev_counts, cat_counts, groups, by_host
Analyzer->>ActionItems: _finalize_action_items(groups, llm_top_n)
ActionItems-->>Analyzer: action_items
Analyzer->>TopDevices: _finalize_top_devices(by_host)
TopDevices-->>Analyzer: top_devices
Analyzer-->>WebApp: AnalysisResult
Analyzer-->>CLI: AnalysisResult
WebApp-->>User: JSON response
CLI-->>User: Printed JSON
Flow diagram for web and CLI streaming inputs with size guardflowchart TD
U[User] -->|HTTP request| W[api_analyze]
U -->|CLI command| C[cmd_analyze]
subgraph WebPath
W --> S[Resolve source]
S -->|directory| D[parse_directory]
S -->|file| F[Check size_mb vs MAX_FILE_MB]
F -->|> MAX_FILE_MB| E413[Return 413 error]
F -->|<= MAX_FILE_MB| PF[parse_file]
D --> G1[events generator]
PF --> G1
G1 --> EC1[Empty check via next]
EC1 -->|no events| E404[Return 404]
EC1 -->|has events| A1[analyze]
end
subgraph CLIPath
C --> B1[Build streams list]
B1 --> CH[itertools.chain.from_iterable]
CH --> EC2[Empty check via next]
EC2 -->|no events| EXIT2[Exit code 2]
EC2 -->|has events| A2[analyze]
end
A1 --> R1[AnalysisResult]
A2 --> R2[AnalysisResult]
R1 --> U
R2 --> U
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- The various aggregation helpers (
_group_actionable,_aggregate_stream,_count_device) rely on loosely-typeddictbuckets with string keys; consider introducing small dataclasses or TypedDicts to make the internal structure explicit and reduce the risk of key/shape mismatches over time. - Constants like the per-group message cap (5) and the global top-k cap (300) are now used in multiple places (heaps, finalization, tests); it may be worth centralizing them as module-level named constants to avoid divergence if these limits are tuned later.
- In
_aggregate_stream, the return tuple is fairly wide and unlabelled at the call site; wrapping these aggregates into a lightweight result object (e.g., dataclass) would improve readability and make future changes to the aggregation outputs less fragile.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The various aggregation helpers (`_group_actionable`, `_aggregate_stream`, `_count_device`) rely on loosely-typed `dict` buckets with string keys; consider introducing small dataclasses or TypedDicts to make the internal structure explicit and reduce the risk of key/shape mismatches over time.
- Constants like the per-group message cap (5) and the global top-k cap (300) are now used in multiple places (heaps, finalization, tests); it may be worth centralizing them as module-level named constants to avoid divergence if these limits are tuned later.
- In `_aggregate_stream`, the return tuple is fairly wide and unlabelled at the call site; wrapping these aggregates into a lightweight result object (e.g., dataclass) would improve readability and make future changes to the aggregation outputs less fragile.
## Individual Comments
### Comment 1
<location path="src/ai_log_analyzer/web/app.py" line_range="85-90" />
<code_context>
+# Web-route cap for single-file analysis. Streaming keeps memory bounded at
+# any size, but classification CPU (~2.7 MB/s) must finish inside the HTTP
+# worker timeout (120s) — bigger files belong on the CLI, which has no timeout.
+MAX_FILE_MB: float = float(os.environ.get("AI_LOG_ANALYZER_MAX_FILE_MB", "200"))
+
+
</code_context>
<issue_to_address>
**suggestion:** File-size check may confuse users around boundary conditions and actual size units
The check divides `st_size` by `1_048_576` but labels the result as `{size_mb:.0f}MB`, which both rounds to an integer and uses MB while actually computing MiB. This can be confusing at the limit (e.g., a ~200.4 MiB file shown as `200MB` but rejected). Consider either displaying one decimal place and/or explicitly labeling the unit as MiB so the message matches the actual threshold more clearly.
Suggested implementation:
```python
size_mb = stat.st_size / 1_048_576
if size_mb > MAX_FILE_MB:
abort(
400,
f"File too large: {size_mb:.1f} MiB (max {MAX_FILE_MB:.1f} MiB). "
"For larger logs, please use the CLI tool instead of the web UI.",
)
```
I assumed:
- The size computation is `size_mb = stat.st_size / 1_048_576`.
- The rejection message is currently `File too large: {size_mb:.0f}MB (max {MAX_FILE_MB:.0f}MB)...`.
If the actual variable names or message text differ, apply the same adjustments wherever the size is:
1. Computed as `st_size / 1_048_576` and formatted with `:.0f`.
2. Labeled with `MB`.
In those places:
- Keep the same variable name (`size_mb` is fine) but change the format specifier from `:.0f` to `:.1f`.
- Change the unit text from `MB` to `MiB` so it matches the binary (MiB) calculation and is clearer at the limit.
</issue_to_address>
### Comment 2
<location path="src/ai_log_analyzer/cli.py" line_range="30-31" />
<code_context>
for path in args.file:
- events.extend(parse_file(path))
+ streams.append(parse_file(path))
if args.stdin:
- events.extend(parse_lines(sys.stdin.readlines()))
+ streams.append(parse_lines(sys.stdin.readlines()))
- if not events:
</code_context>
<issue_to_address>
**issue (performance):** Reading all stdin lines eagerly undermines the streaming guarantees for large inputs
This call to `sys.stdin.readlines()` still buffers all input in memory, which regresses the new bounded-memory streaming behavior for large stdin. To keep stdin consistent with files/FRR logs, pass `sys.stdin` directly to `parse_lines` or iterate over `sys.stdin` instead of using `readlines()` so lines are consumed lazily.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+85
to
90
| MAX_FILE_MB: float = float(os.environ.get("AI_LOG_ANALYZER_MAX_FILE_MB", "200")) | ||
|
|
||
|
|
||
| def _path_allowed(p: Path) -> bool: | ||
| """True when the resolved path sits under one of FILE_ROOTS.""" | ||
| try: |
There was a problem hiding this comment.
suggestion: File-size check may confuse users around boundary conditions and actual size units
The check divides st_size by 1_048_576 but labels the result as {size_mb:.0f}MB, which both rounds to an integer and uses MB while actually computing MiB. This can be confusing at the limit (e.g., a ~200.4 MiB file shown as 200MB but rejected). Consider either displaying one decimal place and/or explicitly labeling the unit as MiB so the message matches the actual threshold more clearly.
Suggested implementation:
size_mb = stat.st_size / 1_048_576
if size_mb > MAX_FILE_MB:
abort(
400,
f"File too large: {size_mb:.1f} MiB (max {MAX_FILE_MB:.1f} MiB). "
"For larger logs, please use the CLI tool instead of the web UI.",
)I assumed:
- The size computation is
size_mb = stat.st_size / 1_048_576. - The rejection message is currently
File too large: {size_mb:.0f}MB (max {MAX_FILE_MB:.0f}MB)....
If the actual variable names or message text differ, apply the same adjustments wherever the size is:
- Computed as
st_size / 1_048_576and formatted with:.0f. - Labeled with
MB.
In those places:
- Keep the same variable name (
size_mbis fine) but change the format specifier from:.0fto:.1f. - Change the unit text from
MBtoMiBso it matches the binary (MiB) calculation and is clearer at the limit.
Comment on lines
30
to
+31
| if args.stdin: | ||
| events.extend(parse_lines(sys.stdin.readlines())) | ||
| streams.append(parse_lines(sys.stdin.readlines())) |
There was a problem hiding this comment.
issue (performance): Reading all stdin lines eagerly undermines the streaming guarantees for large inputs
This call to sys.stdin.readlines() still buffers all input in memory, which regresses the new bounded-memory streaming behavior for large stdin. To keep stdin consistent with files/FRR logs, pass sys.stdin directly to parse_lines or iterate over sys.stdin instead of using readlines() so lines are consumed lazily.
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
1 participant
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
Closes the last verified performance finding: '1GB syslog analyze dies: full in-memory materialization (~6.5GB RSS) + CPU vs 120s gunicorn timeout'.
Approach
One bounded pass (
_aggregate_stream) replaces materialize-everything-then-sort:(timestamp, -seq)— reproducesclassify_events()'s (severity, ts desc, arrival) ordering exactly, asserted by a contract testclassifier.iter_classify()is the new lazy building block;classify_events(),build_action_items(),_extract_top_devices()keep their public contracts. Web file/directory sources and the CLI pass generators end-to-end. The web route gainsAI_LOG_ANALYZER_MAX_FILE_MB(default 200, HTTP 413 over) because streaming fixes memory, not the worker's 120s CPU budget — the CLI streams multi-GB files with no limit.Measured (100MB synthetic syslog, 1.04M events, identical outputs)
Memory is now flat regardless of input size — 1GB stays ~40MB instead of ~6.5GB.
Test plan
ruff checkcleancli.pywas 0% covered)🤖 Generated with Claude Code
Summary by Sourcery
Streamline log analysis to run in bounded memory for any input size while preserving existing behavior and interfaces.
New Features:
Enhancements:
Tests: