Repository navigation
fix: bound pattern-mode memory and fix Kafka read and reduction defects - #25
Merged
Merged
Conversation
Pattern mode kept a result for every Drain cluster it ever created, so max_clusters bounded the template store but not the memory around it. And the fuzzy LSH at enhanced and maximum never dropped an entry. Both grew with the number of distinct lines: 10x the distinct input gave about 10x the tracemalloc peak (28.5 MB against 2.9 MB at standard, 124.5 MB against 11.8 MB at enhanced). - An evicted cluster's result now waits only until the dedup cache can no longer raise its count, then competes for one of max_patterns places. One that loses can never win a place back, so the output is the same as ranking every template ever seen. - The fuzzy index holds the last dedup_cache_size admitted lines, dropping the oldest first, the same window exact dedup keeps. FuzzyDeduplicator takes it as max_entries. - Output is unchanged on 264 reduce runs (11 corpora, 4 modes, 3 levels, default and max_clusters=7). It changes only once more than dedup_cache_size near-unique lines go through fuzzy dedup. - README, architecture and pypi docs now say what bounds pattern-mode memory. Fixes #23
KafkaSource read until every partition had reported EOF, and never cleared an EOF. A topic written to during the read was read past its start-time end, so passes of one reduce saw different messages, and a partition that took new writes after its EOF still counted as drained. Each partition's high watermark is now recorded when it is assigned, the read stops there by offset or by EOF, and messages past it wait for the next read. Read against a Redpanda broker: 2,000 lines plus 50 written mid-read used to return 2,050.
Three consecutive empty polls ended a KafkaSource read as if it were complete, so a slow fetch handed the reducer a partial topic with no sign anything was missing. Against a Redpanda broker a 1 ms poll_timeout returned no lines at all from a 2,000-line topic. Empty polls no longer end a read. It finishes only at the recorded end offsets, and raises KafkaConsumerError once idle_timeout seconds (30 by default) pass with no message while a partition is short of its end. idle_timeout replaces idle_polls. The only caller passing idle_polls was this repo's own integration test, edited here.
KafkaSink.write let a BufferError from a full local queue escape on the first line past the queue limit, and any produce error skipped the flush, so lines already queued could be dropped when the producer was torn down. A full queue now serves delivery reports until there is room, for up to queue_full_timeout seconds (30 by default), then raises KafkaDeliveryError. The flush runs in a finally, and a flush that leaves messages undelivered still raises. Against a Redpanda broker: a 10-message queue used to fail on 2,000 lines, and an oversized line used to leave 5 earlier lines queued.
msg.topic() and msg.partition() can be None, so the drained set took an untyped key, and dropping the PEP 563 import left KafkaSink.__enter__ naming its own class before it existed. Both now type-check under ty.
Fuzzy dedup dropped a near-duplicate without telling anything, so neither it nor its later exact repeats counted. Two lines differing only in whitespace left the pattern at a count of one, and the default min_pattern_occurrences of 2 dropped it from the output. A dropped near-duplicate now adds one to the pattern of the admitted line it matched, and its Occurrence points there for later repeats. An evicted pattern's count is taken as final twice dedup_cache_size lines after its last Occurrence, since a matched near-duplicate can keep feeding it for one more window. This changes output at enhanced and maximum where near-duplicates were present: those patterns now carry their true counts.
auto.offset.reset=earliest applies only to a group with no committed offset, so a KafkaSource on a group_id another consumer had committed for started there and skipped the history before it. Each assigned partition is now set to OFFSET_BEGINNING in on_assign. Against a Redpanda broker, a group committed at offset 50 of 100 used to return the last 50 lines.
LogReducer applied max_memory_gb, max_patterns and keyword overrides with setattr after the config was built, so the 70% of available RAM clamp in BigDialConfig.__post_init__ never saw them: LogReducer(max_memory_gb=999999) kept 999999 and sized file strategies and reservoirs from it. Overrides now go through dataclasses.replace, which re-runs validation and the clamp.
TimestampParser matched an ISO-8601 stamp up to the seconds and dropped any fraction and offset, and the Apache pattern dropped its zone, so 01:00+01:00 and 00:00Z landed in different windows. ISO stamps are now read by fromisoformat with their fraction and offset, the Apache form takes its zone, and aware timestamps are shifted to UTC before windowing. Naive stamps are still read as UTC.
Temporal mode took up to max_patterns timed examples and then up to 100 timeless ones on top, ranked separately, so max_patterns=1 returned three lines and a low-ranked timeless pattern skipped the global selection. Both kinds are now ranked together and cut once at max_patterns.
Anomaly mode gave anomalies max_patterns // 2 and normal context max_patterns // 4, and hybrid split max_patterns // 2 each way, so max_patterns=1 returned nothing in either mode. Anomalies now take half the cap rounded up and the other share is at least one line of what is left. Fewer than 10 unique lines, which the detector cannot score, now fill the cap instead of a quarter of it. At the default caps the budgets are unchanged.
is_severe matched CRITICAL, FATAL or ERROR anywhere in the upper-cased line, so INFO error_count=0 or a line saying no errors counted as severe, bypassed the min-occurrence filter and every output cap, and took a severe-shape slot. A level now has to be a whole word, in any case, so [error], level=FATAL and ERROR: still count. The quiet-lines golden was rebuilt with drain3 0.9.11 through tests/testdata/build_quiet_golden.py, whose severity check now matches the library's. It replays every corpus against the all-lines golden before writing.
With adaptive sizing on, the first batch was a fixed 10,000 rows whatever their size, and the memory watchdog only ran once that batch had been fetched and reduced. With 100 KB rows that first pull is about 1 GB before anything is measured. The first batch is now the 1,000-row floor, and every later one is sized to the byte budget from the rows already seen.
A seeded sample was setseed then random() < fraction on PostgreSQL and rand(seed) < fraction on MySQL, so each row's draw depended on the order the engine scanned it in. A table holding the same rows in reverse order returned a different sample on both engines, which breaks the same-rows-every-pass promise the multi-pass modes rely on. A seeded sample now keeps a row when the first 32 bits of md5(row text + seed) fall under fraction * 2^32: the whole row as text on PostgreSQL, and the query's columns, read once with a LIMIT 0 run, on MySQL. MySQL's from_table predicate hashes its column the same way. build_sample_sql takes the quoted columns as a new optional argument and no longer returns a PostgreSQL setseed statement.
The Docker and Kubernetes recipes in examples/README.md used python:3.12-slim, which cannot install a package that requires 3.14. Both now use python:3.14-slim, and a test holds every image in that file to requires-python.
ty still targeted Python 3.12 after requires-python moved to 3.14. It now targets 3.14. That surfaced two things: an unguarded re.fullmatch in tests/unit/test_examples.py, now asserted first, and the drain3 imports in tests/testdata/build_quiet_golden.py, a reference script run under uvx with drain3 and now excluded from ty.
The ClickHouse tests skipped wherever the docker ClickHouse failed to start, and each run started its own Redpanda. The suite now loads scalo.deployment.test_support.pytest_plugin (scalo>=2.32.4 in the dev group): one ClickHouse and one Kafka per run, shared by every xdist worker, with a database or topics per test, and a failure rather than a skip in CI. Kafka auto-create is off there, so the Kafka tests make single-partition topics through kafka_topics and name groups through kafka_group. PostgreSQL and MySQL stay on testcontainers. scalo 2.32.4 was released today, so it gets a per-package exclude-newer entry, due out on 2026-11-08. The certifi entry that expired on 2026-10-28 is removed. .env.example and CONTRIBUTING.md name the HYPERI_TEST_* settings in place of CLICKHOUSE_* and KAFKA_*.
catinspace-au
changed the title
Change ruff, mypy and vermin targets from Python 3.12 to 3.14 to match requires-python. - ruff drops the quotes from two forward-reference annotations (UP037) and formats two except clauses without brackets (PEP 758). - Annotate the empty list in core.py as list[str]. mypy requires this at 3.12 too. - No compatibility issues at 3.14. vermin 1.8.0 ignores [tool.vermin] in pyproject.toml, so that block only documents the target.
A seeded MySQL SQLSource constructor opened a connection for query column names. Construction failed if the server was unreachable. An invalid fraction raised a connection error instead of a ValueError. The constructor now only validates the request. The first pass reads columns through its query connection and saves the built SQL for later passes.
The eight invariants in docs/architecture.md were a run of bold-lead items. They are now plain numbered lines, one per invariant, unwrapped. Invariant 8 now says the 70% RAM clamp covers LogReducer arguments as well as the config.
|
Released in v3.5.6 -- https://github.com/hyperi-io/logreducer/releases/tag/v3.5.6 |
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
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.
Fix 2 P1 memory leaks and P2 defects in logreducer. Each fix has a regression test. Kafka, PostgreSQL and MySQL tests use real servers.
Fixes #23
Memory
dedup_cache_sizeadmitted lines, about 180 MB at enhanced and 360 MB at maximum.Kafka
idle_timeoutseconds while short of its end raisesKafkaConsumerError.KafkaSink.writewaits for space in a full producer queue, up toqueue_full_timeout, and always flushes.Reduction
error_count=0no longer marks a line severe.max_patternscap.LogReduceroverrides are clamped to 70% of available RAM and validated.reduce_to_targetstarts on a 1,000-row batch until it knows row sizes.Public API changes
KafkaSource(idle_polls=)is nowidle_timeout=(seconds, default 30). Only this repo's tests passed it.KafkaSink(queue_full_timeout=), default 30.FuzzyDeduplicator(max_entries=), whichLogReducersets todedup_cache_size.build_sample_sql(..., columns=). It no longer returns a PostgreSQLsetseedstatement.Output against main
Comparison: 11 corpora, 20,000 lines each, 4 modes, 3 levels, default config.
Tests
scalo.deployment.test_support.pytest_plugin(scalo>=2.32.4, dev group only). Each service has one container per run, shared across xdist workers. Previously skipped ClickHouse tests run.requires-python.