diff --git a/README.md b/README.md index a824b62..902880f 100644 --- a/README.md +++ b/README.md @@ -293,8 +293,9 @@ reducer = logreducer.LogReducer(mode="hybrid", config=cfg) # ...or let logreducer read your env cascade directly: prefixed beats bare. cfg = logreducer.BigDialConfig.from_env("MYAPP", "LOGREDUCER") # MYAPP_MAX_MEMORY_GB wins -# 2. Logging: your app owns the loguru sinks; logreducer registers nothing and -# its records flow through YOUR handlers, formatted by YOUR standard. +# 2. Logging: your app owns the stdlib logging handlers; logreducer attaches +# nothing and its records flow through YOUR handlers, at YOUR level, +# formatted by YOUR standard. logreducer.setup_logging(enable=True, own_sinks=False) ``` @@ -324,7 +325,7 @@ Logging is off by default. Enable it and pick a format via env vars: LOG_LEVEL=DEBUG LOG_FORMAT=json logreducer app.log --log ``` -`LOG_FORMAT=json` emits one JSON object per line for log aggregators; the default is human-readable (coloured in a terminal, plain in CI/containers). Embedding? See `own_sinks=False` above. +Logging is the standard library `logging` module, under the `logreducer` logger. `LOG_FORMAT=json` emits one JSON object per line for log aggregators (keys `timestamp`, `level`, `target`, `function`, `line_number`, `message`, `fields`); the default is human-readable (coloured in a terminal, plain in CI/containers). Embedding? See `own_sinks=False` above. ## Development @@ -342,6 +343,8 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) for the toolchain, test layout (including Copyright 2026 HYPERI PTY LIMITED. +Credits: template mining (`src/logreducer/template_mining.py`) is derived from [Drain3](https://github.com/logpai/Drain3), Copyright (c) 2020-2022 International Business Machines and the Drain3 project contributors, MIT licence. Drain3 builds on the Drain parser in LogPAI's [logparser](https://github.com/logpai/logparser). The module header carries Drain3's licence text in full. + ## Context Depth: [docs/architecture.md](docs/architecture.md). @@ -362,6 +365,7 @@ it is given and drops the rest. | `src/logreducer/sources.py`, `sinks.py` | the two structural protocols that ARE the integration surface, plus `FileSource` and `FileSink` | | `src/logreducer/sql.py`, `clickhouse.py`, `kafka.py` | the optional adapters, one extra each | | `src/logreducer/patterns.py`, `anomaly.py`, `temporal.py` | Drain3 mining and fuzzy dedup, Isolation Forest, time-window grouping | +| `src/logreducer/template_mining.py` | the Drain template miner, derived from Drain3 (credited under License) | | `src/logreducer/memory.py` | `MemoryMonitor`, `BoundedDeduplicator`, the file read strategies | | `src/logreducer/sampling.py`, `target.py` | dialect-aware SQL sampling, and the `reduce_to_target` batch loop | | `src/logreducer/config.py` | `BigDialConfig`, the level presets, the `from_env` cascade | @@ -407,7 +411,7 @@ One declared edge in `dfe-infra/suite.yaml`, and it runs the surprising way. `dfe-stack suite --consumer logreducer` returns an empty edge list: no repo in that graph is something this library depends on. The runtime dependencies -(drain3, psutil, loguru, numpy, scikit-learn, typer) are packages, declared in +(psutil, numpy, scikit-learn, typer) are packages, declared in `pyproject.toml`, which is where they belong. The reason dfe-engine gives for not declaring the dependency -- pin it "once it diff --git a/pyproject.toml b/pyproject.toml index e009003..420695d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,11 +32,9 @@ classifiers = [ ] keywords = ["log", "logging", "reduction", "analysis", "pattern-extraction", "memory-efficient", "streaming", "anomaly-detection"] dependencies = [ - "drain3>=0.9.0", - "psutil>=7.2.0", # memory monitoring (MemoryMonitor) - "loguru>=0.7.3", - "numpy>=2.0.0", - "scikit-learn>=1.5.0", # imports as sklearn; >=1.5 for numpy 2 support + "psutil>=7.2.2", # memory monitoring (MemoryMonitor) + "numpy>=2.5.3", + "scikit-learn>=1.9.1", # imports as sklearn "typer>=0.27.3", # CLI ] @@ -51,13 +49,13 @@ enhanced = [ # wants a ready-made source/sink. Each needs its own running service, so they # are covered by docker-gated integration tests. sql = [ - "sqlalchemy>=2.0.30", # cross-DB server-side-cursor streaming (SQLSource) + "sqlalchemy>=2.1.3", # cross-DB server-side-cursor streaming (SQLSource) ] clickhouse = [ - "clickhouse-connect>=1.0", # native block streaming (ClickHouseSource); 1.0 was breaking + "clickhouse-connect>=1.9.0", # native block streaming (ClickHouseSource) ] kafka = [ - "confluent-kafka>=2.14.0", # librdkafka client (KafkaSource/Sink) + "confluent-kafka>=2.15.1", # librdkafka client (KafkaSource/Sink) ] [project.urls] Homepage = "https://github.com/hyperi-io/logreducer" @@ -142,7 +140,7 @@ line-ending = "auto" # MyPy: KEPT STRICT (logreducer is small and already clean-typed). # ignore_missing_imports only silences third-party libs without stubs -# (drain3, datasketch, ...); our own code stays fully typed via +# (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 @@ -265,9 +263,9 @@ dev = [ "vermin>=1.6.0", # Python version-compat detection # Adapter + integration test deps (TEST ONLY - never in the published # wheel; these mirror the optional extras so the adapters can be exercised). - "sqlalchemy>=2.0.30", # SQLSource against SQLite (real, no server) - "clickhouse-connect>=1.0", # ClickHouseSource (local cluster or docker) - "confluent-kafka>=2.14.0", # KafkaSource/KafkaSink + "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 "psycopg[binary]>=3.2", # PostgreSQL driver (real TABLESAMPLE/setseed sampling tests) "pymysql>=1.1.0", # MySQL driver (real RAND(seed) sampling tests) diff --git a/src/logreducer/anomaly.py b/src/logreducer/anomaly.py index 5683393..2ec83b2 100644 --- a/src/logreducer/anomaly.py +++ b/src/logreducer/anomaly.py @@ -1,6 +1,8 @@ """Anomaly detection utilities.""" -from loguru import logger +import logging + +logger = logging.getLogger(__name__) # Try optional imports try: diff --git a/src/logreducer/config.py b/src/logreducer/config.py index e61469e..f5c7ddd 100644 --- a/src/logreducer/config.py +++ b/src/logreducer/config.py @@ -1,5 +1,6 @@ """Configuration and tuning parameters for LogReducer.""" +import logging import os import types import typing @@ -7,7 +8,8 @@ from enum import Enum import psutil -from loguru import logger + +logger = logging.getLogger(__name__) class ProcessingLevel(Enum): diff --git a/src/logreducer/logging_config.py b/src/logreducer/logging_config.py index 794e0c3..28f3b7e 100644 --- a/src/logreducer/logging_config.py +++ b/src/logreducer/logging_config.py @@ -1,4 +1,4 @@ -"""Logging configuration for LogReducer. +"""Logging configuration for LogReducer, on the standard library ``logging``. Output follows one strict standard so logreducer's logs sit uniformly beside a host application's logs in a shared deployment: RFC 3339 timestamps, a @@ -6,8 +6,8 @@ solarized colours in an interactive terminal, a plain prefix in CI, and one JSON object per line when ``LOG_FORMAT=json``. -logreducer is a library, not a service, so it only ever manages its OWN loguru -handlers (filtered to ``logreducer`` records) and never removes a host +logreducer is a library, not a service, so it only ever attaches handlers to its +OWN ``logreducer`` logger and never touches the root logger or a host application's handlers - embedding logreducer inside another application must not disturb that application's logging. @@ -17,17 +17,15 @@ LOG_OUTPUT=stdout|stderr """ -import contextlib +import json +import logging +import logging.handlers import os import sys +from datetime import UTC, datetime from pathlib import Path -from typing import TYPE_CHECKING -from loguru import logger - -if TYPE_CHECKING: - # Read by get_logger's quoted return annotation, which vulture does not parse. - from loguru import Logger # noqa: V104 +PACKAGE_LOGGER_NAME = "logreducer" # Solarized palette (https://ethanschoonover.com/solarized/) for the console # sink - readable on both light and dark terminal backgrounds. @@ -41,9 +39,29 @@ "red": "#dc322f", } -# Handler ids this module has registered, so re-configuring only tears down our -# own sinks - never the host application's. -_HANDLER_IDS: list[int] = [] +_LEVEL_COLOURS = { + logging.DEBUG: SOLARIZED["base01"], + logging.INFO: SOLARIZED["blue"], + logging.WARNING: SOLARIZED["yellow"], + logging.ERROR: SOLARIZED["orange"], + logging.CRITICAL: SOLARIZED["red"], +} + +# Above CRITICAL, so a disabled logreducer emits nothing, not even through a host's handlers. +_DISABLED_LEVEL = logging.CRITICAL + 1 + +# Rotate the file sink at 10 MB and keep seven rotated files. +_FILE_MAX_BYTES = 10 * 1024 * 1024 +_FILE_BACKUPS = 7 + +# Every attribute a bare LogRecord carries; anything else came from the caller's ``extra=``. +_STANDARD_RECORD_ATTRS = frozenset(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"} + +# Handlers this module has attached, so re-configuring only tears down our own +# sinks - never the host application's. +_HANDLERS: list[logging.Handler] = [] + +logger = logging.getLogger(__name__) def _is_ci() -> bool: @@ -67,34 +85,74 @@ def _is_interactive() -> bool: return "UTF-8" in locale or "UTF8" in locale -def _console_format(colorize: bool, ci: bool) -> str: - """Console format string - solarized colours, or a plain CI prefix.""" - if ci or not colorize: - return "[{level: <8}] {name}:{function}:{line} - {message}" - return ( - f"{{time:YYYY-MM-DDTHH:mm:ss.SSSZZ}} | " - f"{{level: <8}} | " - f"{{name}}:" - f"{{function}}:" - f"{{line}} - " - f"{{message}}" - ) +def _paint(text: str, hex_colour: str) -> str: + """Wrap ``text`` in a 24-bit ANSI foreground colour.""" + red, green, blue = (int(hex_colour[i : i + 2], 16) for i in (1, 3, 5)) + return f"\x1b[38;2;{red};{green};{blue}m{text}\x1b[0m" + +def _local_rfc3339(created: float) -> str: + """RFC 3339 local time with milliseconds and the UTC offset.""" + return datetime.fromtimestamp(created).astimezone().isoformat(timespec="milliseconds") -def _file_format() -> str: - """File format string - ASCII only, RFC 3339 timestamp with bracketed level.""" - return "{time:YYYY-MM-DDTHH:mm:ss.SSSZZ} [{level: <8}] {name}:{function}:{line} - {message}" +class _LineFormatter(logging.Formatter): + """Render ``[time] LEVEL name:function:line - message`` on one line (plus any traceback).""" -def _apply_solarized_levels() -> None: - """Colour the loguru levels with the solarized scheme.""" - logger.level("TRACE", color=f"") - logger.level("DEBUG", color=f"") - logger.level("INFO", color=f"") - logger.level("SUCCESS", color=f"") - logger.level("WARNING", color=f"") - logger.level("ERROR", color=f"") - logger.level("CRITICAL", color=f"") + def __init__(self, *, with_time: bool, colour: bool, ascii_only: bool) -> None: + super().__init__() + self._with_time = with_time + self._colour = colour + self._ascii_only = ascii_only + + def format(self, record: logging.LogRecord) -> str: + """Render one record.""" + level = f"{record.levelname:<8}" + where = f"{record.name}:{record.funcName}:{record.lineno}" + message = record.getMessage() + if record.exc_info: + message = f"{message}\n{self.formatException(record.exc_info)}" + + if self._colour: + level_colour = _LEVEL_COLOURS.get(record.levelno, SOLARIZED["blue"]) + line = ( + f"{_paint(_local_rfc3339(record.created), SOLARIZED['green'])} | " + f"{_paint(level, level_colour)} | {_paint(where, SOLARIZED['cyan'])} - {_paint(message, level_colour)}" + ) + elif self._with_time: + line = f"{_local_rfc3339(record.created)} [{level}] {where} - {message}" + else: + line = f"[{level}] {where} - {message}" + + if self._ascii_only: + return line.encode("ascii", "backslashreplace").decode("ascii") + return line + + +class _JsonFormatter(logging.Formatter): + """Render one flat JSON object per record, keyed like scalo's JSON log lines.""" + + def format(self, record: logging.LogRecord) -> str: + """Render one record.""" + timestamp = datetime.fromtimestamp(record.created, UTC).isoformat(timespec="microseconds") + entry: dict[str, object] = { + "timestamp": timestamp.replace("+00:00", "Z"), + "level": record.levelname, + "target": record.name, + "function": record.funcName, + "line_number": record.lineno, + "message": record.getMessage(), + "fields": {key: value for key, value in record.__dict__.items() if key not in _STANDARD_RECORD_ATTRS}, + } + if record.exc_info: + entry["exception"] = self.formatException(record.exc_info) + return json.dumps(entry, default=str, ensure_ascii=True) + + +def _attach(package_logger: logging.Logger, handler: logging.Handler, formatter: logging.Formatter) -> None: + handler.setFormatter(formatter) + package_logger.addHandler(handler) + _HANDLERS.append(handler) def setup_logging( @@ -107,88 +165,73 @@ def setup_logging( ) -> None: """Configure logreducer's own logging sinks. + Records always propagate to the host application's handlers as well, so a + host that has configured the root logger sees logreducer's records there. + Args: enable: Whether logreducer emits logs at all (default False). - log_file: Optional path for an ASCII, rotating file sink. + log_file: Optional path for an ASCII file sink that rotates at 10 MB + and keeps seven rotated files. log_level: Threshold level (overridden by ``LOG_LEVEL``). log_format: ``text`` (human) or ``json`` (overridden by ``LOG_FORMAT``). console: Also log to the console (stderr, or ``LOG_OUTPUT=stdout``). - own_sinks: When False, register NO handlers - just enable the package - and let logreducer's records flow through the HOST application's - loguru handlers. This is the embedding seam: a host app that owns + own_sinks: When False, attach NO handlers and leave the threshold to the + host - logreducer's records flow to the HOST application's handlers + at the host's level. This is the embedding seam: a host app that owns logging calls ``setup_logging(enable=True, own_sinks=False)`` and logreducer's output lands in the host's sinks, formatted by the host's standard. + + Raises: + ValueError: If the resolved level is not a ``logging`` level name. """ # Env beats caller, so a deployment can retune logging without code changes. log_level = os.environ.get("LOG_LEVEL", log_level).upper() fmt_selector = (os.environ.get("LOG_FORMAT") or log_format or "text").strip().lower() serialize = fmt_selector == "json" - # Tear down ONLY our previously registered handlers. - global _HANDLER_IDS - for handler_id in _HANDLER_IDS: - try: - logger.remove(handler_id) - except ValueError: - pass - _HANDLER_IDS = [] + # Tear down ONLY our previously attached handlers. + package_logger = logging.getLogger(PACKAGE_LOGGER_NAME) + for handler in _HANDLERS: + package_logger.removeHandler(handler) + handler.close() + _HANDLERS.clear() if not enable: - logger.disable("logreducer") + package_logger.setLevel(_DISABLED_LEVEL) return if not own_sinks: - # Host-owned logging: enable the package, register nothing. - logger.enable("logreducer") + package_logger.setLevel(logging.NOTSET) return + package_logger.setLevel(log_level) ci = _is_ci() - stream = sys.stdout if os.environ.get("LOG_OUTPUT") == "stdout" else sys.stderr if console: - # CLI/app mode owns the console: drop loguru's default stderr handler - # (id 0) so our formatted line is not printed twice. Only the pristine - # default is removed - a host app's own handlers (ids > 0) are untouched - # (embedding logreducer only ever goes through the console=False path). - with contextlib.suppress(ValueError): - logger.remove(0) - colorize = not serialize and not ci and _is_interactive() - if colorize: - _apply_solarized_levels() - _HANDLER_IDS.append( - logger.add( - stream, - level=log_level, - format=_console_format(colorize, ci), - colorize=colorize, - serialize=serialize, - filter="logreducer", - ) + stream = sys.stdout if os.environ.get("LOG_OUTPUT") == "stdout" else sys.stderr + colour = not serialize and not ci and _is_interactive() + console_formatter = ( + _JsonFormatter() if serialize else _LineFormatter(with_time=colour, colour=colour, ascii_only=False) ) + _attach(package_logger, logging.StreamHandler(stream), console_formatter) if log_file: try: Path(log_file).parent.mkdir(parents=True, exist_ok=True) - _HANDLER_IDS.append( - logger.add( - log_file, - level=log_level, - format=_file_format(), - rotation="10 MB", - retention="7 days", - encoding="utf-8", - serialize=serialize, - filter="logreducer", - ) + file_handler = logging.handlers.RotatingFileHandler( + log_file, maxBytes=_FILE_MAX_BYTES, backupCount=_FILE_BACKUPS, encoding="utf-8" ) except OSError as exc: # A bad log path should never take the process down. - logger.warning(f"Could not create log file {log_file}: {exc}") - - logger.enable("logreducer") + logger.warning("Could not create log file %s: %s", log_file, exc) + else: + file_formatter = ( + _JsonFormatter() if serialize else _LineFormatter(with_time=True, colour=False, ascii_only=True) + ) + _attach(package_logger, file_handler, file_formatter) -def get_logger(name: str = "logreducer") -> "Logger": - """Return a logger bound for the given logreducer submodule name.""" - return logger.bind(name=name) +def get_logger(name: str = PACKAGE_LOGGER_NAME) -> logging.Logger: + """Return the logger for a logreducer module name (``logreducer.``).""" + return logging.getLogger(name) diff --git a/src/logreducer/patterns.py b/src/logreducer/patterns.py index cfa89a2..fa95ab3 100644 --- a/src/logreducer/patterns.py +++ b/src/logreducer/patterns.py @@ -1,17 +1,17 @@ """Pattern extraction (Drain3 template mining) and fuzzy deduplication.""" +import logging from collections.abc import Iterable, Iterator from dataclasses import dataclass, field from typing import TYPE_CHECKING -from drain3 import TemplateMiner -from drain3.masking import MaskingInstruction -from drain3.template_miner_config import TemplateMinerConfig -from loguru import logger +from .template_mining import MaskingInstruction, TemplateMiner, TemplateMinerConfig if TYPE_CHECKING: from .config import BigDialConfig +logger = logging.getLogger(__name__) + # datasketch powers fuzzy dedup and ships in the optional `enhanced` extra; # without it FuzzyDeduplicator degrades to a pass-through (with a warning). try: @@ -34,10 +34,10 @@ class LogPattern: metadata: dict = field(default_factory=dict) -def _typed_masking_instructions() -> list[MaskingInstruction]: +def _typed_masking_instructions() -> tuple[MaskingInstruction, ...]: """Build the curated masking set for typed template slots. - Drain3 applies these to each line before clustering, so masked values + The miner applies these to each line before clustering, so masked values become literal typed tokens (````, ````, ...) rather than collapsing to the bare ``<*>`` wildcard. Instructions run sequentially over already-masked text, so the specific shapes (IP, UUID, MAC, IPv6) @@ -70,52 +70,47 @@ def _typed_masking_instructions() -> list[MaskingInstruction]: # Integer (drain3's documented NUM example). (r"[-+]?\d+", "NUM"), ] - return [MaskingInstruction(f"{start}(?:{pattern}){end}", name) for pattern, name in shapes] + return tuple(MaskingInstruction(f"{start}(?:{pattern}){end}", name) for pattern, name in shapes) class PatternExtractor: - """Extract patterns using Drain3's online template miner.""" + """Extract patterns using the online template miner (see template_mining).""" def __init__(self, config: "BigDialConfig") -> None: self.config = config self.setup_drain3() def setup_drain3(self) -> None: - """Configure Drain3.""" - drain_config = TemplateMinerConfig() - drain_config.profiling_enabled = False - drain_config.drain_sim_th = self.config.drain_similarity - drain_config.drain_depth = 4 - drain_config.snapshot_interval_minutes = 0 # Disable snapshots - drain_config.snapshot_compress_state = False - # Bound the template store when configured: Drain3 LRU-evicts beyond - # drain_max_clusters, keeping memory flat on high-cardinality logs. - drain_config.drain_max_clusters = self.config.max_clusters + """Configure the template miner.""" # Opt-in typed masking: pre-cluster masking of well-known value shapes # so templates carry typed slots (, , ...) instead of <*>. - if self.config.typed_masking: - drain_config.masking_instructions = _typed_masking_instructions() - self.miner = TemplateMiner(config=drain_config) + masking = _typed_masking_instructions() if self.config.typed_masking else () + drain_config = TemplateMinerConfig( + drain_sim_th=self.config.drain_similarity, + drain_depth=4, + # Bound the template store when configured: the miner LRU-evicts beyond + # drain_max_clusters, keeping memory flat on high-cardinality logs. + drain_max_clusters=self.config.max_clusters, + masking_instructions=masking, + ) + self.miner = TemplateMiner(drain_config) def extract_patterns(self, lines: Iterable[str]) -> list[LogPattern]: """Extract patterns from lines (accepts a stream; only clusters are held). - Feeds each line into Drain3's online miner, so the caller can pass a + Feeds each line into the online miner, so the caller can pass a generator - no need to materialise the whole line set. Memory is bounded by the number of clusters (see ``max_clusters``), not the line count. """ - pattern_map = {} + pattern_map: dict[int, LogPattern] = {} for line in lines: - result = self.miner.add_log_message(line) - cluster_id = result["cluster_id"] - - if cluster_id not in pattern_map: - cluster = self.miner.drain.id_to_cluster[cluster_id] + cluster = self.miner.add_log_message(line) + pattern = pattern_map.get(cluster.cluster_id) + if pattern is None: pattern = LogPattern(template=cluster.get_template(), count=0) - pattern_map[cluster_id] = pattern + pattern_map[cluster.cluster_id] = pattern - pattern = pattern_map[cluster_id] pattern.count += 1 if len(pattern.examples) < self.config.examples_per_pattern: @@ -123,11 +118,10 @@ def extract_patterns(self, lines: Iterable[str]) -> list[LogPattern]: # Later lines widen a cluster's slots to <*>, so the stored template is # re-read once the feed ends; an LRU-evicted cluster keeps its snapshot. - clusters = self.miner.drain.id_to_cluster for cluster_id, pattern in pattern_map.items(): - cluster = clusters.get(cluster_id) - if cluster is not None: - pattern.template = cluster.get_template() + live_cluster = self.miner.get_cluster(cluster_id) + if live_cluster is not None: + pattern.template = live_cluster.get_template() patterns = list(pattern_map.values()) diff --git a/src/logreducer/sql.py b/src/logreducer/sql.py index 474f3aa..2b8e232 100644 --- a/src/logreducer/sql.py +++ b/src/logreducer/sql.py @@ -143,7 +143,7 @@ def from_table( from .sampling import build_table_sample_query owns = isinstance(connectable, str) - engine = create_engine(connectable) if owns else connectable + engine = create_engine(connectable) if isinstance(connectable, str) else connectable preparer = engine.dialect.identifier_preparer query = build_table_sample_query( engine.dialect.name, diff --git a/src/logreducer/template_mining.py b/src/logreducer/template_mining.py new file mode 100644 index 0000000..372c1a6 --- /dev/null +++ b/src/logreducer/template_mining.py @@ -0,0 +1,453 @@ +# Upstream: Drain3, https://github.com/logpai/Drain3 +# +# SPDX-License-Identifier: MIT +# This file implements the Drain algorithm for log parsing. +# Based on https://github.com/logpai/logparser/blob/master/logparser/Drain/Drain.py by LogPAI team +# +# MIT License +# +# Copyright (c) 2020-2022 International Business Machines +# and the Drain3 project contributors. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +# Project: logreducer +# File: src/logreducer/template_mining.py +# Purpose: Online log template mining (the Drain algorithm) with regex pre-masking +# Language: Python +# +# License: Apache-2.0 +# Copyright: (c) 2026 HYPERI PTY LIMITED + +"""Online log template mining with the Drain algorithm. + +Derived from Drain3 at commit 0fb6d6e1828838fa05c8f6fa5cda878e02ea79c7. It keeps +only the streaming path logreducer drives: mask a line, route it through a +fixed-depth prefix tree, and merge it into the most similar cluster of the same +token count, widening differing tokens to ``<*>``. Drain3's copyright and MIT +licence notice are in the header above; the README credits Drain3 and logparser. +""" + +import re +from collections import OrderedDict +from collections.abc import Iterable, Sequence +from dataclasses import dataclass + +# Wildcard a widened template slot carries; masks render as in the same shape. +_PARAM_STR = "<*>" + + +class LogCluster: + """One mined template and the number of lines merged into it. + + Attributes: + log_template_tokens: The template, one token per whitespace-split word. + cluster_id: Sequential id, never reused within a miner. + size: Lines merged into this cluster. + """ + + __slots__ = ("cluster_id", "log_template_tokens", "size") + + def __init__(self, log_template_tokens: Iterable[str], cluster_id: int) -> None: + self.log_template_tokens = tuple(log_template_tokens) + self.cluster_id = cluster_id + self.size = 1 + + def get_template(self) -> str: + """Return the template as one space-joined string.""" + return " ".join(self.log_template_tokens) + + +class ClusterStore: + """Cluster id to cluster map, optionally bounded by least-recently-used eviction. + + Storing and ``touch`` mark a cluster as used; ``get`` and ``in`` only look, so + scoring candidate clusters never changes which one is evicted next. + """ + + __slots__ = ("_clusters", "_maxsize") + + def __init__(self, maxsize: int | None = None) -> None: + """Create an empty store. + + Args: + maxsize: Clusters kept before the least recently used is evicted. + ``None`` keeps every cluster. + + Raises: + ValueError: If ``maxsize`` is below 1. + """ + if maxsize is not None and maxsize < 1: + raise ValueError(f"maxsize must be at least 1, not {maxsize}") + self._maxsize = maxsize + self._clusters: OrderedDict[int, LogCluster] = OrderedDict() + + def __setitem__(self, cluster_id: int, cluster: LogCluster) -> None: + full = self._maxsize is not None and len(self._clusters) >= self._maxsize + if full and cluster_id not in self._clusters: + self._clusters.popitem(last=False) + self._clusters[cluster_id] = cluster + self._clusters.move_to_end(cluster_id) + + def __contains__(self, cluster_id: object) -> bool: + return cluster_id in self._clusters + + def __len__(self) -> int: + return len(self._clusters) + + def get(self, cluster_id: int) -> LogCluster | None: + """Return the cluster without marking it used, or ``None`` if absent or evicted.""" + return self._clusters.get(cluster_id) + + def touch(self, cluster_id: int) -> None: + """Mark a stored cluster as the most recently used.""" + self._clusters.move_to_end(cluster_id) + + def values(self) -> list[LogCluster]: + """Return the stored clusters, least recently used first.""" + return list(self._clusters.values()) + + +class Node: + """A prefix-tree node: child nodes keyed by token, and the clusters stored here.""" + + __slots__ = ("cluster_ids", "key_to_child_node") + + def __init__(self) -> None: + self.key_to_child_node: dict[str, Node] = {} + self.cluster_ids: list[int] = [] + + +def _has_numbers(token: str) -> bool: + return any(char.isdigit() for char in token) + + +class Drain: + """The Drain online clustering algorithm over a fixed-depth prefix tree. + + The first tree level splits lines by token count, the next levels by leading + tokens (a token containing a digit routes to the ``<*>`` child), and each leaf + holds the clusters a line is scored against. + + Attributes: + max_node_depth: Depth of the leaf nodes, counting the token-count level as 1. + sim_th: Minimum share of matching tokens to join an existing cluster. + max_children: Children a tree node holds before new tokens share ``<*>``. + root_node: Root of the prefix tree. + id_to_cluster: Every live cluster, keyed by id. + clusters_counter: The id given to the newest cluster. + """ + + def __init__( + self, + depth: int = 4, + sim_th: float = 0.4, + max_children: int = 100, + max_clusters: int | None = None, + ) -> None: + """Create an empty miner. + + Args: + depth: Tree depth, counting the root and the token-count level. + Depth 4 routes on the first token only. + sim_th: Minimum share of matching tokens to join an existing cluster. + max_children: Children a tree node holds before new tokens share ``<*>``. + max_clusters: Clusters kept before the least recently used is evicted. + ``None`` keeps every cluster. + + Raises: + ValueError: If ``depth`` is below 3 or ``max_clusters`` is below 1. + """ + if depth < 3: + raise ValueError(f"depth must be at least 3, not {depth}") + self.max_node_depth = depth - 2 + self.sim_th = sim_th + self.max_children = max_children + self.root_node = Node() + self.id_to_cluster = ClusterStore(max_clusters) + self.clusters_counter = 0 + + @property + def clusters(self) -> list[LogCluster]: + """Return every live cluster.""" + return self.id_to_cluster.values() + + def add_log_message(self, content: str) -> LogCluster: + """Merge one (already masked) line into its cluster, creating one if none matches. + + Args: + content: The line to cluster. + + Returns: + The cluster the line joined. Its template reflects this line. + """ + content_tokens = content.split() + match_cluster = self.tree_search(content_tokens) + + if match_cluster is None: + self.clusters_counter += 1 + match_cluster = LogCluster(content_tokens, self.clusters_counter) + self.id_to_cluster[match_cluster.cluster_id] = match_cluster + self.add_seq_to_prefix_tree(match_cluster) + return match_cluster + + match_cluster.log_template_tokens = self.create_template(content_tokens, match_cluster.log_template_tokens) + match_cluster.size += 1 + self.id_to_cluster.touch(match_cluster.cluster_id) + return match_cluster + + def tree_search(self, tokens: Sequence[str]) -> LogCluster | None: + """Return the best live cluster for ``tokens`` at or above ``sim_th``, or ``None``. + + Args: + tokens: The line, split on whitespace. + + Returns: + The matching cluster, or ``None`` when the line needs a new one. + """ + token_count = len(tokens) + cur_node = self.root_node.key_to_child_node.get(str(token_count)) + if cur_node is None: + return None + + # Empty lines share the single cluster stored at their token-count node. + if token_count == 0: + return self.id_to_cluster.get(cur_node.cluster_ids[0]) + + cur_node_depth = 1 + for token in tokens: + if cur_node_depth >= self.max_node_depth or cur_node_depth == token_count: + break + child = cur_node.key_to_child_node.get(token) + if child is None: + child = cur_node.key_to_child_node.get(_PARAM_STR) + if child is None: + return None + cur_node = child + cur_node_depth += 1 + + return self.fast_match(cur_node.cluster_ids, tokens) + + def add_seq_to_prefix_tree(self, cluster: LogCluster) -> None: + """Route a new cluster down the tree and store its id at the leaf it reaches. + + Args: + cluster: The cluster just created. + """ + token_count = len(cluster.log_template_tokens) + token_count_str = str(token_count) + if token_count_str not in self.root_node.key_to_child_node: + self.root_node.key_to_child_node[token_count_str] = Node() + cur_node = self.root_node.key_to_child_node[token_count_str] + + if token_count == 0: + cur_node.cluster_ids = [cluster.cluster_id] + return + + current_depth = 1 + for token in cluster.log_template_tokens: + if current_depth >= self.max_node_depth or current_depth >= token_count: + # Drop ids the store has evicted before adding the new one. + live_ids = [cluster_id for cluster_id in cur_node.cluster_ids if cluster_id in self.id_to_cluster] + live_ids.append(cluster.cluster_id) + cur_node.cluster_ids = live_ids + break + + children = cur_node.key_to_child_node + if token in children: + cur_node = children[token] + elif _has_numbers(token): + if _PARAM_STR not in children: + children[_PARAM_STR] = Node() + cur_node = children[_PARAM_STR] + elif _PARAM_STR in children: + if len(children) < self.max_children: + children[token] = Node() + cur_node = children[token] + else: + cur_node = children[_PARAM_STR] + elif len(children) + 1 < self.max_children: + children[token] = Node() + cur_node = children[token] + elif len(children) + 1 == self.max_children: + children[_PARAM_STR] = Node() + cur_node = children[_PARAM_STR] + else: + cur_node = children[_PARAM_STR] + + current_depth += 1 + + def fast_match(self, cluster_ids: Sequence[int], tokens: Sequence[str]) -> LogCluster | None: + """Return the most similar live cluster at or above ``sim_th``, or ``None``. + + Ties on similarity go to the cluster with more ``<*>`` slots, then to the + earlier candidate. + + Args: + cluster_ids: Candidate clusters stored at the leaf the line reached. + tokens: The line, split on whitespace. + + Returns: + The best candidate, or ``None`` if none reaches ``sim_th``. + """ + max_sim = -1.0 + max_param_count = -1 + max_cluster = None + + for cluster_id in cluster_ids: + cluster = self.id_to_cluster.get(cluster_id) + if cluster is None: + continue + cur_sim, param_count = self.get_seq_distance(cluster.log_template_tokens, tokens) + if cur_sim > max_sim or (cur_sim == max_sim and param_count > max_param_count): + max_sim = cur_sim + max_param_count = param_count + max_cluster = cluster + + if max_sim >= self.sim_th: + return max_cluster + return None + + def get_seq_distance(self, template: Sequence[str], tokens: Sequence[str]) -> tuple[float, int]: + """Score a line against a template of the same length. + + ``<*>`` slots count towards neither similarity nor mismatch. + + Args: + template: The cluster template tokens. + tokens: The line tokens. + + Returns: + The share of template tokens the line matches exactly, and the number + of ``<*>`` slots in the template. + """ + if not template: + return 1.0, 0 + + sim_tokens = 0 + param_count = 0 + for template_token, token in zip(template, tokens, strict=True): + if template_token == _PARAM_STR: + param_count += 1 + elif template_token == token: + sim_tokens += 1 + + return sim_tokens / len(template), param_count + + def create_template(self, tokens: Sequence[str], template: Sequence[str]) -> tuple[str, ...]: + """Widen every template token the line does not match to ``<*>``. + + Args: + tokens: The line tokens. + template: The current template tokens. + + Returns: + The merged template. + """ + return tuple( + template_token if token == template_token else _PARAM_STR + for token, template_token in zip(tokens, template, strict=True) + ) + + +class MaskingInstruction: + """Replace every regex match in a line with ```` before clustering. + + Attributes: + regex: The compiled pattern. + mask_with: Name the match renders as, inside angle brackets. + """ + + __slots__ = ("mask_with", "regex") + + def __init__(self, pattern: str, mask_with: str) -> None: + """Compile a masking rule. + + Args: + pattern: Regular expression for the values to mask. + mask_with: Name the match renders as, inside angle brackets. + """ + self.regex = re.compile(pattern) + self.mask_with = mask_with + + def mask(self, content: str) -> str: + """Return ``content`` with every match replaced by ````.""" + return self.regex.sub(f"<{self.mask_with}>", content) + + +@dataclass(frozen=True, slots=True) +class TemplateMinerConfig: + """Tuning for a :class:`TemplateMiner`. + + Attributes: + drain_sim_th: Minimum share of matching tokens to join an existing cluster. + drain_depth: Prefix-tree depth, counting the root and the token-count level. + drain_max_children: Children a tree node holds before new tokens share ``<*>``. + drain_max_clusters: Clusters kept before the least recently used is evicted. + ``None`` keeps every cluster. + masking_instructions: Masks applied to each line in order, before clustering. + """ + + drain_sim_th: float = 0.4 + drain_depth: int = 4 + drain_max_children: int = 100 + drain_max_clusters: int | None = None + masking_instructions: tuple[MaskingInstruction, ...] = () + + +class TemplateMiner: + """Mask each line, then cluster it with :class:`Drain`. + + Attributes: + config: The tuning this miner was built with. + drain: The clustering state. + """ + + def __init__(self, config: TemplateMinerConfig | None = None) -> None: + """Create an empty miner. + + Args: + config: Tuning; ``None`` uses the :class:`TemplateMinerConfig` defaults. + """ + self.config = config if config is not None else TemplateMinerConfig() + self.drain = Drain( + depth=self.config.drain_depth, + sim_th=self.config.drain_sim_th, + max_children=self.config.drain_max_children, + max_clusters=self.config.drain_max_clusters, + ) + + def add_log_message(self, log_message: str) -> LogCluster: + """Mask one line and merge it into its cluster. + + Args: + log_message: The raw line. + + Returns: + The cluster the line joined. Its template keeps widening as later + lines join, so read it again once the feed ends. + """ + masked = log_message + for instruction in self.config.masking_instructions: + masked = instruction.mask(masked) + return self.drain.add_log_message(masked) + + def get_cluster(self, cluster_id: int) -> LogCluster | None: + """Return a live cluster without marking it used, or ``None`` once evicted.""" + return self.drain.id_to_cluster.get(cluster_id) diff --git a/src/logreducer/temporal.py b/src/logreducer/temporal.py index 17d54b6..1bf1d7c 100644 --- a/src/logreducer/temporal.py +++ b/src/logreducer/temporal.py @@ -6,7 +6,7 @@ from datetime import datetime from typing import Any -from drain3 import TemplateMiner +from .template_mining import TemplateMiner @dataclass @@ -102,21 +102,16 @@ def _extract_temporal_patterns(self, entries: list[LogEntry]) -> list[dict]: # Process each window for window_key, window_entries in windows.items(): if window_key not in self.miners_by_window: - from drain3.template_miner_config import TemplateMinerConfig - - config = TemplateMinerConfig() - config.snapshot_interval_minutes = 0 # Disable snapshots - config.snapshot_compress_state = False - self.miners_by_window[window_key] = TemplateMiner(config=config) + self.miners_by_window[window_key] = TemplateMiner() miner = self.miners_by_window[window_key] - cluster_ids = [miner.add_log_message(entry.raw_line)["cluster_id"] for entry in window_entries] + clusters = [miner.add_log_message(entry.raw_line) for entry in window_entries] # Templates are read once the window is fed, after later lines have widened their slots. - for entry, cluster_id in zip(window_entries, cluster_ids, strict=True): + for entry, cluster in zip(window_entries, clusters, strict=True): patterns.append( { - "template": miner.drain.id_to_cluster[cluster_id].get_template(), + "template": cluster.get_template(), "window": window_key, "count": 1, "example": entry.raw_line, @@ -127,21 +122,16 @@ def _extract_temporal_patterns(self, entries: list[LogEntry]) -> list[dict]: def _extract_timeless_patterns(self, entries: list[LogEntry]) -> list[dict]: """Extract patterns without timestamps.""" - from drain3.template_miner_config import TemplateMinerConfig - - config = TemplateMinerConfig() - config.snapshot_interval_minutes = 0 # Disable snapshots - config.snapshot_compress_state = False - miner = TemplateMiner(config=config) - cluster_ids = [miner.add_log_message(entry.raw_line)["cluster_id"] for entry in entries] + miner = TemplateMiner() + clusters = [miner.add_log_message(entry.raw_line) for entry in entries] # Templates are read once every entry is fed, after later lines have widened their slots. return [ { - "template": miner.drain.id_to_cluster[cluster_id].get_template(), + "template": cluster.get_template(), "example": entry.raw_line, } - for entry, cluster_id in zip(entries, cluster_ids, strict=True) + for entry, cluster in zip(entries, clusters, strict=True) ] def _analyze_distribution(self, entries: list[LogEntry]) -> dict: diff --git a/tests/integration/test_template_equivalence.py b/tests/integration/test_template_equivalence.py new file mode 100644 index 0000000..11d8e13 --- /dev/null +++ b/tests/integration/test_template_equivalence.py @@ -0,0 +1,57 @@ +"""Template mining reproduces the drain3 0.9.11 output it replaced, on real corpora. + +``tests/testdata/template_mining_golden.json.gz`` holds what drain3 0.9.11 +produced for the first ``limit`` lines of every committed corpus under each +miner configuration: the per-line cluster assignment (as a sha256 digest), the +number of live clusters, and the templates and counts ``extract_patterns`` +returned, in order. Server-free; no mocks. +""" + +import gzip +import hashlib +import json +from typing import Any + +import pytest + +from logreducer.config import BigDialConfig +from logreducer.patterns import PatternExtractor +from tests.integration import corpora + +pytestmark = pytest.mark.integration + +GOLDEN_PATH = corpora.TESTDATA_DIR / "template_mining_golden.json.gz" +GOLDEN: dict[str, Any] = json.loads(gzip.decompress(GOLDEN_PATH.read_bytes())) +CASES = [(name, label) for name in GOLDEN["results"] for label in GOLDEN["configs"]] + + +def _config(label: str) -> BigDialConfig: + # Keep every cluster, so the comparison covers the whole template set. + return BigDialConfig(min_pattern_occurrences=1, max_patterns=10**9, **GOLDEN["configs"][label]) + + +def test_golden_covers_every_corpus(): + assert sorted(GOLDEN["results"]) == sorted(corpora.dataset_names()) + + +@pytest.mark.parametrize(("name", "label"), CASES) +def test_cluster_assignment_matches_drain3(name, label): + expected = GOLDEN["results"][name][label] + lines = corpora.read_all(name, limit=GOLDEN["limit"]) + assert len(lines) == expected["lines"] + + miner = PatternExtractor(_config(label)).miner + ids = [miner.add_log_message(line).cluster_id for line in lines] + + assert hashlib.sha256(",".join(map(str, ids)).encode()).hexdigest() == expected["assignments_sha256"] + assert len(miner.drain.id_to_cluster) == expected["live_clusters"] + + +@pytest.mark.parametrize(("name", "label"), CASES) +def test_extracted_patterns_match_drain3(name, label): + expected = GOLDEN["results"][name][label] + lines = corpora.read_all(name, limit=GOLDEN["limit"]) + + patterns = PatternExtractor(_config(label)).extract_patterns(lines) + + assert [[p.template, p.count] for p in patterns] == expected["patterns"] diff --git a/tests/testdata/README.md b/tests/testdata/README.md index 18c7878..28114e8 100644 --- a/tests/testdata/README.md +++ b/tests/testdata/README.md @@ -30,6 +30,10 @@ template counts from a handful to well over a thousand. | `elastic_nginx_json` | structured | one JSON object per line | | `secrepo_auth` | security | auth.log SSH failures | +## Template-mining golden + +`template_mining_golden.json.gz` is the template-mining output drain3 0.9.11 produced on the first 3000 lines of each corpus, under five miner configurations. `tests/integration/test_template_equivalence.py` holds `logreducer.template_mining` to it. Rebuilding a corpus invalidates that corpus's entries. + ## PII cleansing `build.py` rewrites IPs (-> `10.x`), IPv6 (-> `2001:db8::`), emails diff --git a/tests/testdata/template_mining_golden.json.gz b/tests/testdata/template_mining_golden.json.gz new file mode 100644 index 0000000..5609660 Binary files /dev/null and b/tests/testdata/template_mining_golden.json.gz differ diff --git a/tests/unit/test_embedding_seams.py b/tests/unit/test_embedding_seams.py index 12e10b8..f466450 100644 --- a/tests/unit/test_embedding_seams.py +++ b/tests/unit/test_embedding_seams.py @@ -75,25 +75,33 @@ def test_unset_fields_keep_defaults(self): class TestHostOwnedLogging: - def test_own_sinks_false_flows_records_to_host_handlers(self): - """The embedding seam: logreducer registers nothing; the host's own - loguru handler receives logreducer's records (which are filtered by - the caller's module name, so they must originate inside the package - - drive them via a real reduce()).""" - from loguru import logger + def test_own_sinks_false_flows_records_to_host_handlers(self, caplog, tmp_path): + """The embedding seam: logreducer attaches nothing; the host's own root + handler (caplog's, at the host's INFO level) receives logreducer's + records - driven via a real reduce() so they originate in the package.""" + import logging from logreducer import logging_config + log_file = tmp_path / "ignored-when-host-owned.log" reducer = LogReducer(level="standard", mode="pattern", enable_logging=True) - setup_logging(enable=True, own_sinks=False, console=True, log_file="ignored-when-host-owned.log") + setup_logging(enable=True, own_sinks=False, console=True, log_file=str(log_file)) try: - assert logging_config._HANDLER_IDS == [] # nothing registered by us - captured: list[str] = [] - host_id = logger.add(lambda m: captured.append(str(m)), filter="logreducer") - try: - reducer.reduce(["ERROR a", "ERROR a", "INFO b"]) - finally: - logger.remove(host_id) - assert captured, "logreducer records did not reach the host handler" + assert logging_config._HANDLERS == [] # nothing attached by us + assert logging.getLogger("logreducer").handlers == [] + caplog.set_level(logging.INFO) + reducer.reduce(["ERROR a", "ERROR a", "INFO b"]) + ours = [r for r in caplog.records if r.name.startswith("logreducer.")] + assert ours, "logreducer records did not reach the host handler" + assert not log_file.exists() finally: setup_logging(enable=False) # restore the library default + + def test_disabled_logging_reaches_no_host_handler(self, caplog): + """Disabled (the default) means silent, even with a host handler at DEBUG.""" + import logging + + caplog.set_level(logging.DEBUG) + reducer = LogReducer(level="standard", mode="pattern") + reducer.reduce(["ERROR a", "ERROR a", "INFO b"]) + assert [r for r in caplog.records if r.name.startswith("logreducer")] == [] diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 24dcf3c..02f3333 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -6,6 +6,8 @@ """ import json +import logging +from datetime import datetime import pytest @@ -38,9 +40,35 @@ def test_json_log_format(tmp_path, small_log_file, monkeypatch): reducer.process_file(str(small_log_file)) setup_logging(enable=False) - first_line = log_file.read_text().splitlines()[0] - record = json.loads(first_line) # loguru serialize=True -> one JSON object per line - assert "record" in record + lines = log_file.read_text().splitlines() + records = [json.loads(line) for line in lines] # one JSON object per line + assert records + first = records[0] + assert set(first) >= {"timestamp", "level", "target", "function", "line_number", "message", "fields"} + assert first["target"].startswith("logreducer.") + assert first["timestamp"].endswith("Z") + + +def test_text_log_file_is_ascii_with_rfc3339_time(tmp_path, small_log_file): + log_file = tmp_path / "run.log" + reducer = LogReducer(level="standard", enable_logging=True, log_file=str(log_file), log_level="INFO") + reducer.process_file(str(small_log_file)) + setup_logging(enable=False) + + first_line = log_file.read_text(encoding="utf-8").splitlines()[0] + first_line.encode("ascii") # raises if the file sink wrote a non-ASCII byte + timestamp, rest = first_line.split(" ", 1) + assert datetime.fromisoformat(timestamp).tzinfo is not None # offset-aware RFC 3339 + assert rest.startswith("[INFO ] logreducer.core:") + + +def test_file_sink_escapes_non_ascii(tmp_path): + log_file = tmp_path / "unicode.log" + setup_logging(enable=True, log_file=str(log_file), log_level="INFO") + logging.getLogger("logreducer.test").info("caf" + chr(0xE9)) + setup_logging(enable=False) + + assert log_file.read_text(encoding="ascii").rstrip().endswith("caf\\xe9") def test_disabled_logging_writes_nothing(tmp_path, small_log_file): diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py index 74ffcf8..1924c8e 100644 --- a/tests/unit/test_patterns.py +++ b/tests/unit/test_patterns.py @@ -56,44 +56,36 @@ def test_initialization_with_config(self): assert extractor.config == config assert hasattr(extractor, "miner") - @patch("logreducer.patterns.TemplateMiner") - def test_extract_patterns_empty_input(self, mock_template_miner): - """Test pattern extraction with empty input""" - config = BigDialConfig() - extractor = PatternExtractor(config) - - result = extractor.extract_patterns([]) + def test_extract_patterns_empty_input(self): + """Empty input yields no patterns.""" + extractor = PatternExtractor(BigDialConfig()) - assert result == [] + assert extractor.extract_patterns([]) == [] - @patch("logreducer.patterns.TemplateMiner") - def test_extract_patterns_basic(self, mock_template_miner): - """Test basic pattern extraction""" + def test_extract_patterns_basic(self): + """Lines differing in one token merge into one widened pattern.""" config = BigDialConfig(max_patterns=10, examples_per_pattern=2) - - # Mock the miner and cluster - mock_cluster = Mock() - mock_cluster.get_template.return_value = "User <*> logged in" - - mock_miner = Mock() - mock_miner.add_log_message.return_value = {"cluster_id": 1} - mock_miner.drain.id_to_cluster = {1: mock_cluster} - mock_template_miner.return_value = mock_miner - extractor = PatternExtractor(config) - extractor.miner = mock_miner - lines = ["User alice logged in", "User bob logged in", "User charlie logged in"] result = extractor.extract_patterns(lines) - # Should have processed lines - assert mock_miner.add_log_message.call_count == len(lines) + assert len(result) == 1 + assert result[0].template == "User <*> logged in" + assert result[0].count == 3 + assert result[0].examples == lines[:2] - # Should return patterns - assert isinstance(result, list) - if len(result) > 0: - assert isinstance(result[0], LogPattern) + def test_evicted_cluster_keeps_its_first_template(self): + """A cluster LRU-evicted mid-feed keeps the template it had when first seen.""" + extractor = PatternExtractor(BigDialConfig(max_clusters=1, min_pattern_occurrences=1)) + lines = ["alpha one x", "alpha two x", "beta gamma delta epsilon"] + + result = extractor.extract_patterns(lines) + + templates = {p.template: p.count for p in result} + # "alpha" was evicted by "beta" before the feed ended, so its widened + # "alpha <*> x" is never re-read; it keeps the first line's template. + assert templates == {"alpha one x": 2, "beta gamma delta epsilon": 1} def test_filter_by_occurrence(self): """Test filtering patterns by minimum occurrence""" @@ -293,7 +285,7 @@ def test_priority_reads_the_final_template(self): def test_default_configures_no_masking_instructions(self): """Off by default: the miner carries zero masking instructions.""" extractor = PatternExtractor(BigDialConfig()) - assert list(extractor.miner.masker.masking_instructions) == [] + assert extractor.miner.config.masking_instructions == () def test_typed_masking_ip(self): """IP-bearing lines yield a typed slot, no bare wildcard.""" diff --git a/tests/unit/test_template_mining.py b/tests/unit/test_template_mining.py new file mode 100644 index 0000000..93da184 --- /dev/null +++ b/tests/unit/test_template_mining.py @@ -0,0 +1,199 @@ +"""Unit tests for the template miner (logreducer.template_mining). + +Real miner, no mocks. The cross-check against the drain3 output it replaced +lives in tests/integration/test_template_equivalence.py. +""" + +import importlib.metadata +import re +import subprocess +import sys + +import pytest + +from logreducer.template_mining import ( + ClusterStore, + Drain, + LogCluster, + MaskingInstruction, + TemplateMiner, + TemplateMinerConfig, +) + +# Packages the miner replaced; none may be imported or declared any more. +REPLACED = {"drain3", "jsonpickle", "cachetools", "loguru"} + + +class TestClustering: + def test_differing_token_widens_to_wildcard(self): + miner = TemplateMiner() + first = miner.add_log_message("user alice logged in") + second = miner.add_log_message("user bob logged in") + + assert first is second + assert second.get_template() == "user <*> logged in" + assert second.size == 2 + + def test_below_similarity_threshold_starts_a_new_cluster(self): + # One of four tokens matches: 0.25 is below the default 0.4. + miner = TemplateMiner() + first = miner.add_log_message("disk full on sda") + second = miner.add_log_message("disk ok for now") + + assert first.cluster_id != second.cluster_id + + def test_lower_threshold_merges_the_same_lines(self): + miner = TemplateMiner(TemplateMinerConfig(drain_sim_th=0.2)) + first = miner.add_log_message("disk full on sda") + second = miner.add_log_message("disk ok for now") + + assert first is second + assert second.get_template() == "disk <*> <*> <*>" + + def test_token_count_partitions_clusters(self): + miner = TemplateMiner() + short = miner.add_log_message("job done") + long = miner.add_log_message("job done quickly") + + assert short.cluster_id != long.cluster_id + + def test_empty_and_blank_lines_share_one_empty_cluster(self): + miner = TemplateMiner() + empty = miner.add_log_message("") + blank = miner.add_log_message(" \t ") + + assert empty is blank + assert empty.get_template() == "" + assert empty.size == 2 + + def test_whitespace_runs_split_like_single_spaces(self): + miner = TemplateMiner() + first = miner.add_log_message(" a b c ") + second = miner.add_log_message("a b c") + + assert first is second + assert second.get_template() == "a b c" + + def test_cluster_ids_are_sequential(self): + miner = TemplateMiner() + ids = [miner.add_log_message(line).cluster_id for line in ("a", "b c", "d e f")] + + assert ids == [1, 2, 3] + + def test_full_node_routes_new_first_tokens_to_the_wildcard_child(self): + # max_children=3: alpha and beta get their own nodes, the third slot + # becomes <*>, and gamma and delta both land under it and merge. + miner = TemplateMiner(TemplateMinerConfig(drain_max_children=3)) + clusters = [miner.add_log_message(f"{word} x y") for word in ("alpha", "beta", "gamma", "delta")] + + assert len({c.cluster_id for c in clusters}) == 3 + assert clusters[2] is clusters[3] + assert clusters[3].get_template() == "<*> x y" + + def test_numeric_first_token_routes_to_the_wildcard_child(self): + miner = TemplateMiner() + miner.add_log_message("42 requests served") + + first_layer = miner.drain.root_node.key_to_child_node["3"] + assert list(first_layer.key_to_child_node) == ["<*>"] + + def test_depth_below_three_is_rejected(self): + with pytest.raises(ValueError, match="depth must be at least 3"): + Drain(depth=2) + + +class TestMasking: + def test_mask_renders_as_named_slot(self): + miner = TemplateMiner(TemplateMinerConfig(masking_instructions=(MaskingInstruction(r"\d+", "NUM"),))) + cluster = miner.add_log_message("took 15 ms") + + assert cluster.get_template() == "took ms" + + def test_masks_apply_in_order(self): + # The IP mask runs first, so the NUM mask never sees the octets. + instructions = ( + MaskingInstruction(r"\d{1,3}(?:\.\d{1,3}){3}", "IP"), + MaskingInstruction(r"\d+", "NUM"), + ) + miner = TemplateMiner(TemplateMinerConfig(masking_instructions=instructions)) + + assert miner.add_log_message("from 10.0.0.1 port 22").get_template() == "from port " + + +class TestClusterStore: + def _cluster(self, cluster_id: int) -> LogCluster: + return LogCluster(["t"], cluster_id) + + def test_unbounded_store_keeps_everything(self): + store = ClusterStore() + for cluster_id in range(1, 101): + store[cluster_id] = self._cluster(cluster_id) + + assert len(store) == 100 + + def test_evicts_least_recently_used(self): + store = ClusterStore(maxsize=2) + store[1] = self._cluster(1) + store[2] = self._cluster(2) + store.touch(1) + store[3] = self._cluster(3) + + assert 2 not in store + assert 1 in store + assert 3 in store + + def test_get_does_not_refresh_recency(self): + store = ClusterStore(maxsize=2) + store[1] = self._cluster(1) + store[2] = self._cluster(2) + assert store.get(1) is not None + store[3] = self._cluster(3) + + assert store.get(1) is None + assert store.get(2) is not None + + def test_replacing_a_stored_id_evicts_nothing(self): + store = ClusterStore(maxsize=2) + store[1] = self._cluster(1) + store[2] = self._cluster(2) + store[1] = self._cluster(1) + + assert len(store) == 2 + + @pytest.mark.parametrize("maxsize", [0, -1]) + def test_maxsize_below_one_is_rejected(self, maxsize): + with pytest.raises(ValueError, match="maxsize must be at least 1"): + ClusterStore(maxsize=maxsize) + + def test_miner_get_cluster_returns_none_once_evicted(self): + miner = TemplateMiner(TemplateMinerConfig(drain_max_clusters=1)) + first = miner.add_log_message("a b") + second = miner.add_log_message("c d e") + + assert miner.get_cluster(first.cluster_id) is None + assert miner.get_cluster(second.cluster_id) is second + + +class TestNoReplacedDependency: + def test_importing_logreducer_loads_none_of_the_replaced_packages(self): + code = ( + "import sys, logreducer, logreducer.core, logreducer.patterns, logreducer.temporal\n" + "print(sorted(m for m in sys.modules if m.split('.')[0] in " + f"{sorted(REPLACED)!r}))" + ) + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=True, + ) + + assert result.stdout.strip() == "[]" + + def test_distribution_declares_none_of_the_replaced_packages(self): + requirements = importlib.metadata.requires("logreducer") or [] + names = {re.split(r"[\s;<>=!~\[(]", req, maxsplit=1)[0].lower() for req in requirements} + + assert names.isdisjoint(REPLACED) diff --git a/uv.lock b/uv.lock index e8384df..c18dfa8 100644 --- a/uv.lock +++ b/uv.lock @@ -125,15 +125,6 @@ filecache = [ { name = "filelock" }, ] -[[package]] -name = "cachetools" -version = "4.2.1" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/74/17/5735dd9f015f03d2d928ea108f3c02075b784ceed05d32a98e7e44ddd114/cachetools-4.2.1.tar.gz", hash = "sha256:f469e29e7aa4cff64d8de4aad95ce76de8ea1125a16c68e0d93f65c3c3dc92e9", size = 24753, upload-time = "2021-01-24T22:40:13.746Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/bb/72/8df2e0dc991f1a1d2c6869404e7622e8ee50d80bff357dbb57c3df70305b/cachetools-4.2.1-py3-none-any.whl", hash = "sha256:1d9d5f567be80f7c07d765e21b814326d78c61eb0c3a637dffc0e5d1796cb2e2", size = 12003, upload-time = "2021-01-24T22:40:11.795Z" }, -] - [[package]] name = "certifi" version = "2026.7.22" @@ -566,16 +557,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/75/23/529140fe1aab80fc6992f93a706deec709140a6397439139a054e1515c45/docker-7.2.0-py3-none-any.whl", hash = "sha256:a3f45fdeb9165e2d25d9a1d02ddf3bc70fb572cf5ebbf9b58558c22caf29b71f", size = 148775, upload-time = "2026-07-09T14:53:45.224Z" }, ] -[[package]] -name = "drain3" -version = "0.9.11" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "cachetools" }, - { name = "jsonpickle" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/dc/83/4da2d3a11b5e0edf1a4f4c0c2dd42126d2eb1f31c733967edd3dfac1af94/drain3-0.9.11.tar.gz", hash = "sha256:9ab4b1407fad74f56554ae371ef019c3c7985861631f4bab46a0e92585125f75", size = 27960, upload-time = "2022-07-17T06:40:11.433Z" } - [[package]] name = "filelock" version = "4.0.12" @@ -615,15 +596,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/18/53/84099323c2ec4be98d935f63c033ac4151ee83836ca1050ede3b3aadf155/joblib-1.6.0-py3-none-any.whl", hash = "sha256:3dbbf9f6e4b592a2357b854608e980fe6390d131d7a82f011a377ef2ebef7aba", size = 306115, upload-time = "2026-08-31T09:39:02.298Z" }, ] -[[package]] -name = "jsonpickle" -version = "1.5.1" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/af/62/31ef0b58050a3731b079af69932104a9443bff07fe2b9564c161e3ec4348/jsonpickle-1.5.1.tar.gz", hash = "sha256:060f97096559d1b86aa16cac2f4ea5f7b6da0c15d8a4de150b78013a886f9a51", size = 109560, upload-time = "2021-01-31T05:57:15.037Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/77/a7/c2f527ddce3155ae9e008385963c2325cbfd52969f8b38efa2723e2af4af/jsonpickle-1.5.1-py2.py3-none-any.whl", hash = "sha256:8eb8323f0e12cb40687f0445e2115d8165901e20ac670add55bb53a95c68c0e5", size = 37124, upload-time = "2021-01-31T05:57:12.256Z" }, -] - [[package]] name = "librt" version = "0.16.0" @@ -711,8 +683,6 @@ name = "logreducer" version = "3.4.0" source = { editable = "." } dependencies = [ - { name = "drain3" }, - { name = "loguru" }, { name = "numpy" }, { name = "psutil" }, { name = "scikit-learn" }, @@ -756,15 +726,13 @@ dev = [ [package.metadata] requires-dist = [ - { name = "clickhouse-connect", marker = "extra == 'clickhouse'", specifier = ">=1.0" }, - { name = "confluent-kafka", marker = "extra == 'kafka'", specifier = ">=2.14.0" }, + { name = "clickhouse-connect", marker = "extra == 'clickhouse'", specifier = ">=1.9.0" }, + { name = "confluent-kafka", marker = "extra == 'kafka'", specifier = ">=2.15.1" }, { name = "datasketch", marker = "extra == 'enhanced'", specifier = ">=2.0.0" }, - { name = "drain3", specifier = ">=0.9.0" }, - { name = "loguru", specifier = ">=0.7.3" }, - { name = "numpy", specifier = ">=2.0.0" }, - { name = "psutil", specifier = ">=7.2.0" }, - { name = "scikit-learn", specifier = ">=1.5.0" }, - { name = "sqlalchemy", marker = "extra == 'sql'", specifier = ">=2.0.30" }, + { name = "numpy", specifier = ">=2.5.3" }, + { name = "psutil", specifier = ">=7.2.2" }, + { name = "scikit-learn", specifier = ">=1.9.1" }, + { name = "sqlalchemy", marker = "extra == 'sql'", specifier = ">=2.1.3" }, { name = "typer", specifier = ">=0.27.3" }, { name = "xxhash", marker = "extra == 'enhanced'", specifier = ">=4.0.1" }, ] @@ -773,8 +741,8 @@ provides-extras = ["enhanced", "sql", "clickhouse", "kafka"] [package.metadata.requires-dev] dev = [ { name = "bandit", extras = ["toml"], specifier = ">=1.7.0" }, - { name = "clickhouse-connect", specifier = ">=1.0" }, - { name = "confluent-kafka", specifier = ">=2.14.0" }, + { name = "clickhouse-connect", specifier = ">=1.9.0" }, + { name = "confluent-kafka", specifier = ">=2.15.1" }, { name = "mypy", specifier = ">=2.0" }, { name = "pip-audit", specifier = ">=2.6.0" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.2" }, @@ -783,26 +751,13 @@ dev = [ { name = "pytest-cov", specifier = ">=7.0.0" }, { name = "python-dotenv", specifier = ">=1.0.0" }, { name = "ruff", specifier = ">=0.16.10" }, - { name = "sqlalchemy", specifier = ">=2.0.30" }, + { name = "sqlalchemy", specifier = ">=2.1.3" }, { name = "testcontainers", extras = ["clickhouse", "kafka", "postgres", "mysql"], specifier = ">=4.15.0" }, { name = "ty", specifier = ">=0.0.34" }, { name = "vermin", specifier = ">=1.6.0" }, { name = "vulture", specifier = ">=2.16" }, ] -[[package]] -name = "loguru" -version = "0.7.3" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "colorama", marker = "sys_platform == 'win32'" }, - { name = "win32-setctime", marker = "sys_platform == 'win32'" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/3a/05/a1dae3dffd1116099471c643b8924f5aa6524411dc6c63fdae648c4f1aca/loguru-0.7.3.tar.gz", hash = "sha256:19480589e77d47b8d85b2c827ad95d49bf31b0dcde16593892eb51dd18706eb6", size = 63559, upload-time = "2024-12-06T11:20:56.608Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/0c/29/0348de65b8cc732daa3e33e67806420b2ae89bdce2b04af740289c5c6c8c/loguru-0.7.3-py3-none-any.whl", hash = "sha256:31a33c10c8e1e10422bfd431aeb5d351c7cf7fa671e3c4df004162264b28220c", size = 61595, upload-time = "2024-12-06T11:20:54.538Z" }, -] - [[package]] name = "lz4" version = "4.4.5" @@ -1702,15 +1657,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f5/be/f935130312330614811dae2ea9df3f395f6d63889eb6c2e68c14507152ee/vulture-2.16-py3-none-any.whl", hash = "sha256:6e0f1c312cef1c87856957e5c2ca9608834a7c794c2180477f30bf0e4cc58eee", size = 26993, upload-time = "2026-03-25T14:41:26.21Z" }, ] -[[package]] -name = "win32-setctime" -version = "1.2.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/b3/8f/705086c9d734d3b663af0e9bb3d4de6578d08f46b1b101c2442fd9aecaa2/win32_setctime-1.2.0.tar.gz", hash = "sha256:ae1fdf948f5640aae05c511ade119313fb6a30d7eabe25fef9764dca5873c4c0", size = 4867, upload-time = "2024-12-07T15:28:28.314Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/e1/07/c6fe3ad3e685340704d314d765b7912993bcb8dc198f0e7a89382d37974b/win32_setctime-1.2.0-py3-none-any.whl", hash = "sha256:95d644c4e708aba81dc3704a116d8cbc974d70b3bdb8be1d150e36be6e9d1390", size = 4083, upload-time = "2024-12-07T15:28:26.465Z" }, -] - [[package]] name = "wrapt" version = "2.5.0"