Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
2b26344
fix: bound evicted patterns and fuzzy index memory
catinspace-au Nov 1, 2026
02b05b8
docs: state the fuzzy index memory per preset
catinspace-au Nov 1, 2026
567895e
fix: stop a Kafka read at the end offsets it started with
catinspace-au Nov 1, 2026
c247214
fix: raise when a Kafka read stalls short of its end
catinspace-au Nov 1, 2026
c467e52
fix: retry a full Kafka producer queue and always flush
catinspace-au Nov 1, 2026
c5a8eea
fix: type the Kafka partition key and the sink context
catinspace-au Nov 1, 2026
774cf6a
fix: count fuzzy-dropped lines towards the pattern they matched
catinspace-au Nov 1, 2026
b1098b4
fix: rewind every Kafka read to the earliest offset
catinspace-au Nov 1, 2026
89aba44
fix: clamp and validate LogReducer config overrides
catinspace-au Nov 1, 2026
9a75743
fix: window temporal timestamps by instant, not wall time
catinspace-au Nov 1, 2026
54b01d5
fix: rank temporal and timeless patterns under one cap
catinspace-au Nov 1, 2026
dd64546
fix: keep anomaly and hybrid output under a small cap
catinspace-au Nov 1, 2026
ea5b55f
fix: match a severity level as a whole word
catinspace-au Nov 1, 2026
5eeb559
fix: start reduce_to_target on a 1,000-row probe batch
catinspace-au Nov 1, 2026
f1bf8bf
fix: key a seeded SQL sample on a hash of the row
catinspace-au Nov 1, 2026
7f7bf9d
docs: run the deployment recipes on Python 3.14
catinspace-au Nov 1, 2026
15f3678
fix: type-check against the Python 3.14 floor
catinspace-au Nov 1, 2026
6e2e4ce
test: run ClickHouse and Kafka tests on scalo's shared services
catinspace-au Nov 1, 2026
96e8710
fix: target Python 3.14 in lint and type checks
catinspace-au Nov 1, 2026
8e80afb
fix: read seeded MySQL sample columns on first read
catinspace-au Nov 1, 2026
e78fa85
docs: list the architecture invariants without bold leads
catinspace-au Nov 1, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 23 additions & 38 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -2,48 +2,33 @@
#
# .env is GITIGNORED and must never be committed (it can hold a real
# credential). Everything here is OPTIONAL: it is read only by the
# integration tests (tests/conftest.py loads .env via python-dotenv) and by
# the logging runtime. The library itself needs none of it.
# integration tests and by the logging runtime. The library itself needs none of it.
#
# Optional env-prefix cascade (scalo convention): set LOGREDUCER_ENV_PREFIX to
# read <PREFIX>_<NAME> before the bare name - e.g. LOGREDUCER_ENV_PREFIX="DFE"
# makes DFE_CLICKHOUSE_HOST win over CLICKHOUSE_HOST, so a .env written for
# another project drops in without renaming.
# LOGREDUCER_ENV_PREFIX="DFE"
#
# Test service selection is env-first, then docker:
# 1. if the vars below point at a reachable service, tests use it;
# 2. otherwise a throwaway docker container is started (ClickHouse, Redpanda
# for Kafka, PostgreSQL, MySQL) and cleaned up at the end;
# 3. if neither is available the dependent test skips.
# CI sets none of these -> it always uses docker. Tests create uniquely-named
# tables/topics and drop them afterwards, so pointing at a shared/persistent
# instance is safe.
# ClickHouse and Kafka for the integration tests come from scalo's shared test
# services: one container each per run, or an always-present server named by the
# HYPERI_TEST_* settings below when it answers (never in CI or under the full
# tier). Without Docker the test skips locally and fails in CI. Each test makes
# its own database or topics and drops them afterwards.

# --- ClickHouse (integration tests) -----------------------------------------
# CLICKHOUSE_HOST="localhost"
# CLICKHOUSE_PORT="8123"
# CLICKHOUSE_USER="default"
# CLICKHOUSE_PASSWORD=""
# CLICKHOUSE_SECURE="false" # true = TLS
# CLICKHOUSE_VERIFY="true" # false = skip TLS certificate verification
# --- ClickHouse pet (integration tests) -------------------------------------
# HYPERI_TEST_CLICKHOUSE_HOST="localhost"
# HYPERI_TEST_CLICKHOUSE_HTTP_PORT="8123"
# HYPERI_TEST_CLICKHOUSE_USER="default"
# HYPERI_TEST_CLICKHOUSE_PASSWORD=""
# HYPERI_TEST_CLICKHOUSE_CLUSTER="default"

# --- Kafka (integration tests) ----------------------------------------------
# A bare bootstrap is a plain (no-auth) broker. Add KAFKA_SECURITY_PROTOCOL to
# reach an authenticated broker (SASL_SSL/SASL_PLAINTEXT); the SASL_* + SSL vars
# are then merged into the librdkafka client config.
# KAFKA_BOOTSTRAP_SERVERS="localhost:9092"
# KAFKA_SECURITY_PROTOCOL="SASL_SSL" # e.g. SASL_SSL | SASL_PLAINTEXT
# KAFKA_SASL_MECHANISM="SCRAM-SHA-512"
# KAFKA_SASL_USERNAME="user"
# KAFKA_SASL_PASSWORD=""
# KAFKA_SSL_CA_LOCATION="/path/to/ca.pem" # CA bundle for TLS verification
# KAFKA_SSL_VERIFY="true" # false = skip TLS cert verification
# --- Kafka pet (integration tests) ------------------------------------------
# HYPERI_TEST_KAFKA_BOOTSTRAP="localhost:9092"
# HYPERI_TEST_KAFKA_SECURITY_PROTOCOL="SASL_SSL"
# HYPERI_TEST_KAFKA_SASL_MECHANISM="SCRAM-SHA-512"
# HYPERI_TEST_KAFKA_USER="user"
# HYPERI_TEST_KAFKA_PASSWORD=""
# HYPERI_TEST_KAFKA_CA_FILE="/path/to/ca.pem"
# HYPERI_TEST_KAFKA_TOPIC_PREFIX="test-"

# --- Docker test containers (optional) --------------------------------------
# Keep the throwaway docker services RUNNING after the run (default: stop +
# remove). Handy for fast re-runs - export the printed endpoint as the vars
# above so the next run takes the env-first path.
# --- PostgreSQL and MySQL test containers (optional) ------------------------
# Keep the throwaway PostgreSQL and MySQL containers RUNNING after the run
# (default: stop + remove).
# LOGREDUCER_KEEP_CONTAINERS="1"

# --- Logging (runtime, optional) ---------------------------------------------
Expand Down
19 changes: 2 additions & 17 deletions CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -37,23 +37,8 @@ uv build # build wheel + sdist
internal code). `tests/integration/` - marked `integration`, run against
real services. `tests/testdata/` holds gzipped, PII-cleansed slices of REAL
public log datasets (see its README for provenance and the rebuild tool).
- Integration tests find their services env-first, then docker, then skip:

```mermaid
flowchart TD
ENV{"env vars set?<br/>(CLICKHOUSE_* /<br/>KAFKA_BOOTSTRAP_SERVERS)"} -->|yes| PROBE{"endpoint<br/>reachable?"}
PROBE -->|yes| USE(["use the configured service<br/>(unique table/topic names,<br/>dropped afterwards)"])
PROBE -->|"no (warns why)"| DOCKER
ENV -->|no| DOCKER{"Docker<br/>available?"}
DOCKER -->|yes| TC(["throwaway testcontainer:<br/>ClickHouse / Redpanda /<br/>PostgreSQL / MySQL"])
TC --> STOP(["stopped at session end<br/>(LOGREDUCER_KEEP_CONTAINERS=1<br/>keeps it + prints the endpoint)"])
DOCKER -->|no| SKIP([pytest.skip])
```

- Copy `.env.example` to `.env` to point tests at existing services. CI sets
no env vars, so it always takes the docker path.
- Every test that touches a shared service creates uniquely-named tables and
topics and cleans them up, so runs never collide.
- ClickHouse and Kafka come from scalo's shared test services (`scalo.deployment.test_support.pytest_plugin`): one container each per run, shared by every xdist worker, or the always-present server the `HYPERI_TEST_*` settings in `.env.example` name. Each test gets its own database or topics, dropped afterwards. Without Docker those tests skip locally and fail in CI.
- PostgreSQL and MySQL are throwaway testcontainers, stopped at the end of the session unless `LOGREDUCER_KEEP_CONTAINERS=1`, and skip without Docker.

## Commit messages

Expand Down
13 changes: 7 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ It ships as a CLI and as a Python library with an IO-agnostic core.
## What it gives you

- A `logreducer` command that reduces a file, a SQL or ClickHouse query, or a Kafka topic, straight from the shell.
- Pattern mode streams through a bounded dedup cache and template store, so memory grows with the number of distinct templates, not with the line count.
- Pattern mode streams through a bounded dedup cache, fuzzy index and template store, so memory does not grow with the line count or the number of distinct lines.
- Four modes: pattern (Drain3 template mining), anomaly (Isolation Forest), temporal (one example per template per time window) and hybrid.
- ERROR, FATAL and CRITICAL lines always come through, one per distinct shape and ahead of the reduced output, in every mode and however a file is read. A SQL or ClickHouse source samples in the database first, so only the lines it returns are seen.
- Lines naming ERROR, FATAL or CRITICAL as a whole word, in any case, always come through, one per distinct shape and ahead of the reduced output, in every mode and however a file is read. A SQL or ClickHouse source samples in the database first, so only the lines it returns are seen.
- A core with no IO dependencies: anything re-iterable that yields `str` lines is a source, and anything with `write(lines)` is a sink.
- Seeded SQL sampling and ClickHouse `SAMPLE` for tables too big to scan in full.
- Adapters for SQL, ClickHouse and Kafka as extras, so you install only the one you use.
Expand Down Expand Up @@ -88,7 +88,7 @@ reduced = reducer.reduce(lines)

## How it works

Pattern mode streams: lines flow through dedup and template mining as one generator pipeline, so the unique-line set is never materialised. Memory grows with the number of distinct templates, since a result is kept for every template seen, not with the line count.
Pattern mode streams: lines flow through dedup and template mining as one generator pipeline, so the unique-line set is never materialised. A template the store evicts keeps its place in the ranking only while it can still make the `max_patterns` cut, so memory is set by `max_clusters`, `max_patterns` and `dedup_cache_size`, not by the input.

```mermaid
flowchart LR
Expand Down Expand Up @@ -223,10 +223,11 @@ logreducer --dsn postgresql://user@host/db --query "SELECT msg FROM logs" \

## Bounding memory

Pattern mode streams, and its memory grows with distinct templates, not lines. Temporal mode holds the whole source in memory, and anomaly mode holds its unique lines. The knobs:
Pattern mode streams, and its memory is fixed by the knobs below, not by lines or distinct templates. Temporal mode holds the whole source in memory, and anomaly mode holds its unique lines. The knobs:

- `max_memory_gb` - the overall cap (also sizes file read strategy and the watchdog). Presets: 0.5 / 1 / 2 GB by level; values above 70% of available RAM are clamped with a warning.
- `max_clusters` - LRU-bound the Drain3 template store (10,000 by default, `None` for unbounded). Mining time grows with the square of the number of templates a line is compared with, so a lower value also speeds up high-cardinality sources.
- `dedup_cache_size` - the lines exact dedup remembers, and at enhanced and maximum the lines the fuzzy index holds. Both drop the oldest first, so a repeat or near-repeat older than the window comes through again.
- `max_clusters` - LRU-bound the Drain3 template store (10,000 by default, `None` for unbounded, which also keeps every evicted template's result). Mining time grows with the square of the number of templates a line is compared with, so a lower value also speeds up high-cardinality sources.
- `anomaly_max_rows` - reservoir-cap the rows fed to anomaly detection (off by default; trades anomaly recall for a bounded matrix).

```python
Expand Down Expand Up @@ -307,7 +308,7 @@ reducer = LogReducer(
level="enhanced",
mode="hybrid",
max_memory_gb=1.0, # memory ceiling (clamped to 70% of available RAM)
dedup_cache_size=100000, # bounded dedup cache
dedup_cache_size=100000, # bounded dedup cache and fuzzy index
drain_similarity=0.4, # pattern similarity threshold
fuzzy_threshold=0.8, # fuzzy-dedup threshold (enhanced/maximum)
anomaly_contamination=0.1, # expected anomaly fraction
Expand Down
33 changes: 13 additions & 20 deletions docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -66,8 +66,11 @@ than pass one, and the reducer would be comparing two populations.

Pattern mode streams end to end. Exact dedup, optional fuzzy dedup and Drain3
mining run as one generator pipeline, so the unique-line set is never collected
into a list. Peak memory is the bounded dedup cache plus the Drain3 template
store, and neither grows with the number of unique lines the source has.
into a list. Peak memory is the bounded dedup cache, the fuzzy index (the same `dedup_cache_size` window), the Drain3 template store and the results of evicted templates, and none of them grows with the number of unique lines the source has.

An evicted template's result is kept only while it can still make the `max_patterns` cut. Its count stops changing once the dedup cache has dropped the lines that fed it, so from then on it competes for one of `max_patterns` places, and a result that loses one can never win it back. That keeps the output identical to ranking every template ever seen.

The fuzzy index measured about 1.8 KB per entry, so the preset windows hold about 180 MB at enhanced (100K lines) and 360 MB at maximum (200K), inside their 1 GB and 2 GB budgets.

Drain3 compares each line with every template stored at the leaf it reaches, so
time grows with the square of the template count when lines share no tokens.
Expand Down Expand Up @@ -128,24 +131,14 @@ Three seams, none of which needs the library to know anything about the host.

## Invariants

1. **`__iter__` returns a fresh iterator.** Checked at the entry point, not
documented and hoped for.
2. **The core imports no IO library.** An adapter's dependency is an extra, and
the core never imports the adapter modules.
3. **Analysis state is per run.** `_reset_components()` rebuilds the dedup
seen-set, both Drain3 miners and the fuzzy LSH, so one `LogReducer` is
reusable across calls and hybrid's two passes cannot poison each other.
Without it every line reads as already-seen and the result is empty.
4. **An unknown config keyword raises.** A silently dropped override is tuning
that quietly does nothing, which is worse than a failure.
5. **The row-to-line convention is shared.** The database adapters take the FIRST
column, skip NULLs and skip blank lines, matching `FileSource`, so the same
data reduces identically whichever source carried it.
6. **Every sample is seeded.** See above.
7. **Config enums serialise as their `.value`**, so metadata survives
`json.dumps` and no `OutputFormat.LINE`-style repr leaks into output.
8. **A memory ceiling above 70% of available RAM is clamped with a warning**
rather than accepted.
1. `__iter__` returns a fresh iterator, checked at the entry point rather than documented and hoped for.
2. The core imports no IO library. An adapter's dependency is an extra, and the core never imports the adapter modules.
3. Analysis state is per run. `_reset_components()` rebuilds the dedup seen-set, both Drain3 miners and the fuzzy LSH, so one `LogReducer` is reusable across calls and hybrid's two passes cannot poison each other. Without it every line reads as already-seen and the result is empty.
4. An unknown config keyword raises. A silently dropped override is tuning that quietly does nothing, which is worse than a failure.
5. The row-to-line convention is shared. The database adapters take the FIRST column, skip NULLs and skip blank lines, matching `FileSource`, so the same data reduces identically whichever source carried it.
6. Every sample is seeded. See above.
7. Config enums serialise as their `.value`, so metadata survives `json.dumps` and no `OutputFormat.LINE`-style repr leaks into output.
8. A memory ceiling above 70% of available RAM is clamped with a warning rather than accepted, whether it comes from the config or a `LogReducer` argument.

### Where the numbers are approximate

Expand Down
6 changes: 3 additions & 3 deletions docs/pypi.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,10 @@ It ships as a CLI and as a Python library with an IO-agnostic core.
## What it gives you

- A `logreducer` command that reduces a file, a SQL or ClickHouse query, or a Kafka topic, straight from the shell.
- Pattern mode streams through a bounded dedup cache and template store, so memory grows with the number of distinct templates, not with the line count.
- Pattern mode streams through a bounded dedup cache, fuzzy index and template store, so memory does not grow with the line count or the number of distinct lines.
- A memory budget that decides how a file is read: whole, in chunks, or reservoir-sampled once it passes five times the budget.
- Four modes: pattern (Drain3 template mining), anomaly (Isolation Forest), temporal (one example per template per time window) and hybrid.
- ERROR, FATAL and CRITICAL lines always come through, one per distinct shape and ahead of the reduced output, in every mode and however a file is read. A SQL or ClickHouse source samples in the database first, so only the lines it returns are seen.
- Lines naming ERROR, FATAL or CRITICAL as a whole word, in any case, always come through, one per distinct shape and ahead of the reduced output, in every mode and however a file is read. A SQL or ClickHouse source samples in the database first, so only the lines it returns are seen.
- A core with no IO dependencies: anything re-iterable that yields `str` lines is a source, and anything with `write(lines)` is a sink.
- Seeded SQL sampling and ClickHouse `SAMPLE` for tables too big to scan in full.
- `reduce_to_target` to collect about N representative lines from a large source, holding roughly one batch in memory.
Expand Down Expand Up @@ -48,7 +48,7 @@ From the shell it is `logreducer app.log -o reduced.log`. `level` is `standard`,

| Extra | Adds |
|---|---|
| `enhanced` | Fuzzy dedup with MinHash LSH, whose index grows with near-unique lines, and faster hashing |
| `enhanced` | Fuzzy dedup with MinHash LSH, whose index holds the last `dedup_cache_size` near-unique lines, and faster hashing |
| `sql` | `SQLSource` over SQLAlchemy, with your own DBAPI driver |
| `clickhouse` | `ClickHouseSource` over clickhouse-connect |
| `kafka` | `KafkaSource` and `KafkaSink` over confluent-kafka |
Expand Down
4 changes: 2 additions & 2 deletions examples/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ sample log.
### Docker container

```dockerfile
FROM python:3.12-slim
FROM python:3.14-slim

RUN pip install logreducer

Expand All @@ -38,7 +38,7 @@ spec:
spec:
containers:
- name: logreducer
image: python:3.12-slim
image: python:3.14-slim
command: ["sh", "-c"]
args:
- |
Expand Down
21 changes: 12 additions & 9 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"Topic :: System :: Logging",
"Topic :: System :: Systems Administration",

Check notice on line 28 in pyproject.toml

View workflow job for this annotation

GitHub Actions / ci / Quality

oss-no-trailing-whitespace

trailing whitespace on line 28
"Topic :: Software Development :: Libraries :: Python Modules",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
Expand Down Expand Up @@ -74,7 +74,7 @@

# Ruff: lint + format (replaces black + flake8 + isort). Line length 120.
[tool.ruff]
target-version = "py312"
target-version = "py314"
line-length = 120
indent-width = 4
src = ["src"]
Expand Down Expand Up @@ -144,7 +144,7 @@
# (datasketch, ...); our own code stays fully typed via
# disallow_untyped_defs below.
[tool.mypy]
python_version = "3.12" # Held below requires-python with ruff; see target-version
python_version = "3.14"
ignore_missing_imports = true
warn_return_any = true
warn_unused_configs = true
Expand All @@ -171,6 +171,7 @@
"slow: marks tests as slow (deselect with '-m \"not slow\"')",
"integration: marks tests as integration tests",
]
shared_service_repo = "logreducer"

# Coverage configuration
[tool.coverage.run]
Expand Down Expand Up @@ -218,9 +219,9 @@
# version published in the last 7 days. A CVE fix bypasses it through a
# temporary per-package exclude-newer-package entry, not by lowering this.
exclude-newer = "7 days"
# certifi PYSEC-2026-4320 is fixed in 2026.10.19. A per-package entry replaces the
# global span rather than widening it, so remove this one on 2026-10-28.
exclude-newer-package = { certifi = "2026-10-21T00:00:00Z" }
# scalo 2.32.4 carries the shared test services the suite needs, released 2026-11-01. A
# per-package entry replaces the global span rather than widening it, so remove this one on 2026-11-08.
exclude-newer-package = { scalo = "2026-11-01T06:00:00Z" }
override-dependencies = []

[[tool.uv.index]]
Expand All @@ -230,10 +231,11 @@

# ty: Astral type checker, runs alongside mypy.
[tool.ty.environment]
python-version = "3.12"
python-version = "3.14"

[tool.ty.src]
exclude = ["examples/", "docs/", "scripts/"]
# build_quiet_golden.py runs under uvx with drain3, which is never a dependency here.
exclude = ["examples/", "docs/", "scripts/", "tests/testdata/build_quiet_golden.py"]

# Below 80 vulture reports the library's public API and attributes set on
# third-party config objects, which buries a real leftover.
Expand All @@ -242,7 +244,7 @@

# vermin: verify no syntax/features newer than the minimum Python.
[tool.vermin]
targets = ["3.12"]
targets = ["3.14"]
no_tips = true

[tool.bandit.assert_used]
Expand Down Expand Up @@ -273,7 +275,8 @@
"sqlalchemy>=2.1.3", # SQLSource against SQLite (real, no server)
"clickhouse-connect>=1.9.0", # ClickHouseSource (local cluster or docker)
"confluent-kafka>=2.15.1", # KafkaSource/KafkaSink
"testcontainers[clickhouse,kafka,postgres,mysql]>=4.15.0", # docker fallback; modules under testcontainers.community
"scalo>=2.32.4", # the run's shared ClickHouse and Kafka (scalo.deployment.test_support)
"testcontainers[postgres,mysql]>=4.15.0", # PostgreSQL and MySQL, which scalo does not provide
"psycopg[binary]>=3.2", # PostgreSQL driver (real TABLESAMPLE/setseed sampling tests)
"pymysql>=1.1.0", # MySQL driver (real RAND(seed) sampling tests)
"python-dotenv>=1.0.0", # load the local .env (CLICKHOUSE_*) in tests
Expand Down
5 changes: 4 additions & 1 deletion src/logreducer/anomaly.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@

logger = logging.getLogger(__name__)

# Fewer lines than this are returned unscored, all of them normal.
MIN_SCORED_LINES = 10

# Try optional imports
try:
from sklearn.ensemble import IsolationForest
Expand All @@ -30,7 +33,7 @@ def detect_anomalies(self, lines: list[str]) -> tuple[list[str], list[str]]:
Returns:
(anomalous_lines, normal_lines)
"""
if not self.enabled or len(lines) < 10:
if not self.enabled or len(lines) < MIN_SCORED_LINES:
return [], lines

# Vectorize
Expand Down
Loading
Loading