diff --git a/README.md b/README.md index 297aa2e..a4ac487 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,15 @@ flowchart LR Supported formats include ISO-8601 application logs, syslog, nginx access logs, and generic level-prefixed lines. Detection is an explainable statistical heuristic; it is not a trained model and no accuracy claim is made without a labeled evaluation corpus. +## Evidence-backed reasoning + +The optional `--explain` flag turns existing detector output into concise, user-facing evidence statements. Each statement identifies its direct or statistical basis: parsed error level, message-length z-score with its configured threshold, or observed error count/rate in a complete analysis window. LogSight does not infer an incident root cause, use an LLM, send logs externally, or report a model-confidence score. + +```bash +logsight analyze application.log --window 200 --spike-threshold 0.20 --explain +cat application.log | logsight stdin --explain +``` + ## Quick start ```bash diff --git a/logsight/analyzer.py b/logsight/analyzer.py index 2ef6ee4..867b10a 100644 --- a/logsight/analyzer.py +++ b/logsight/analyzer.py @@ -26,11 +26,33 @@ def error_rate(self) -> float: return self.error_count / self.total if self.total else 0.0 +@dataclass(frozen=True) +class AnomalyEvidence: + """Detection reason for one anomalous log entry.""" + + entry: LogEntry + reasons: tuple[str, ...] + message_length_zscore: float | None = None + + +@dataclass(frozen=True) +class ErrorRateSpike: + """Observed error rate for one complete analysis window.""" + + start: int + end: int + error_count: int + total: int + error_rate: float + threshold: float + + @dataclass class AnomalyReport: """Result of an anomaly scan over a sequence of log entries.""" anomalies: list[LogEntry] = field(default_factory=list) + evidence: list[AnomalyEvidence] = field(default_factory=list) stats: WindowStats = field(default_factory=WindowStats) zscore_threshold: float = 2.5 @@ -105,21 +127,31 @@ def detect_anomalies( std = pstdev(lengths) anomalous: list[LogEntry] = [] + evidence: list[AnomalyEvidence] = [] seen_ids: set[int] = set() for entry in entries: - flagged = False + reasons: list[str] = [] + zscore: float | None = None if flag_errors and entry.is_error: - flagged = True + reasons.append("error_level") if std > 0: - z = abs(len(entry.message) - mean) / std - if z > zscore_threshold: - flagged = True - if flagged and id(entry) not in seen_ids: + zscore = abs(len(entry.message) - mean) / std + if zscore > zscore_threshold: + reasons.append("message_length_zscore") + if reasons and id(entry) not in seen_ids: anomalous.append(entry) + evidence.append( + AnomalyEvidence( + entry=entry, + reasons=tuple(reasons), + message_length_zscore=zscore, + ) + ) seen_ids.add(id(entry)) report.anomalies = anomalous + report.evidence = evidence return report @@ -148,12 +180,42 @@ def error_rate_spike( if not 0 <= spike_threshold <= 1: raise ValueError("spike_threshold must be between 0 and 1") - spike_starts: list[int] = [] - n = len(entries) - for start in range(0, n - window_size + 1, window_size): + return [ + spike.start + for spike in error_rate_spike_details( + entries, + window_size=window_size, + spike_threshold=spike_threshold, + ) + ] + + +def error_rate_spike_details( + entries: Sequence[LogEntry], + window_size: int = 100, + spike_threshold: float = 0.25, +) -> list[ErrorRateSpike]: + """Return the measured evidence for every complete error-rate spike window.""" + + if window_size <= 0: + raise ValueError("window_size must be greater than zero") + if not 0 <= spike_threshold <= 1: + raise ValueError("spike_threshold must be between 0 and 1") + + spikes: list[ErrorRateSpike] = [] + for start in range(0, len(entries) - window_size + 1, window_size): window = entries[start : start + window_size] - errors = sum(1 for e in window if e.is_error) - rate = errors / window_size - if rate >= spike_threshold: - spike_starts.append(start) - return spike_starts + errors = sum(1 for entry in window if entry.is_error) + error_rate = errors / window_size + if error_rate >= spike_threshold: + spikes.append( + ErrorRateSpike( + start=start, + end=start + window_size - 1, + error_count=errors, + total=window_size, + error_rate=error_rate, + threshold=spike_threshold, + ) + ) + return spikes diff --git a/logsight/cli.py b/logsight/cli.py index 7693939..2f4d62b 100644 --- a/logsight/cli.py +++ b/logsight/cli.py @@ -6,10 +6,12 @@ import click from rich.console import Console +from rich.markup import escape from rich.table import Table -from logsight.analyzer import AnomalyReport, detect_anomalies, error_rate_spike +from logsight.analyzer import AnomalyReport, detect_anomalies, error_rate_spike_details from logsight.parser import parse_file, parse_lines +from logsight.reasoning import EvidenceExplanation, explain_report console = Console() @@ -29,7 +31,7 @@ def _print_report(report: AnomalyReport, show_anomalies: bool) -> None: table.add_column("Count", justify="right", style="cyan", no_wrap=True) table.add_column("Message") for msg, cnt in stats.top_messages: - table.add_row(str(cnt), msg) + table.add_row(str(cnt), escape(msg)) console.print(table) if show_anomalies and report.has_anomalies: @@ -37,7 +39,7 @@ def _print_report(report: AnomalyReport, show_anomalies: bool) -> None: for entry in report.anomalies[:20]: level_style = "red" if entry.is_error else "yellow" console.print( - f" [[{level_style}]{entry.level.value}[/{level_style}]] {entry.message[:200]}" + f" [[{level_style}]{entry.level.value}[/{level_style}]] {escape(entry.message[:200])}" ) if len(report.anomalies) > 20: console.print(f" … and {len(report.anomalies) - 20} more.") @@ -45,6 +47,20 @@ def _print_report(report: AnomalyReport, show_anomalies: bool) -> None: console.print("\n[bold green]No anomalies detected.[/bold green]") +def _print_explanations(explanations: list[EvidenceExplanation]) -> None: + """Render concise detector evidence without making causal claims.""" + + if not explanations: + return + console.print("\n[bold]Evidence-backed findings[/bold]") + for explanation in explanations: + evidence = ", ".join(explanation.evidence) + console.print( + f" [{explanation.evidence_strength}] {escape(explanation.summary)} " + f"([dim]{escape(evidence)}[/dim])" + ) + + @click.group() @click.version_option() def main() -> None: @@ -83,12 +99,19 @@ def main() -> None: show_default=True, help="Error-rate fraction that constitutes a spike.", ) +@click.option( + "--explain", + is_flag=True, + default=False, + help="Show evidence-backed detector explanations.", +) def analyze_cmd( logfile: str, threshold: float, no_anomalies: bool, window: int, spike_threshold: float, + explain: bool, ) -> None: """Analyze LOGFILE and report anomalies.""" try: @@ -104,12 +127,18 @@ def analyze_cmd( report = detect_anomalies(entries, zscore_threshold=threshold) _print_report(report, show_anomalies=not no_anomalies) - spikes = error_rate_spike(entries, window_size=window, spike_threshold=spike_threshold) + spikes = error_rate_spike_details( + entries, + window_size=window, + spike_threshold=spike_threshold, + ) if spikes: console.print( f"\n[bold red]Error-rate spikes at windows starting at lines: " - f"{', '.join(str(s) for s in spikes)}[/bold red]" + f"{', '.join(str(spike.start) for spike in spikes)}[/bold red]" ) + if explain: + _print_explanations(explain_report(report, spikes)) @main.command("stdin") @@ -121,7 +150,13 @@ def analyze_cmd( show_default=True, help="Z-score threshold for anomaly detection.", ) -def stdin_cmd(threshold: float) -> None: +@click.option( + "--explain", + is_flag=True, + default=False, + help="Show evidence-backed detector explanations.", +) +def stdin_cmd(threshold: float, explain: bool) -> None: """Read log lines from stdin and report anomalies.""" lines = sys.stdin.readlines() entries = parse_lines(lines) @@ -130,6 +165,8 @@ def stdin_cmd(threshold: float) -> None: return report = detect_anomalies(entries, zscore_threshold=threshold) _print_report(report, show_anomalies=True) + if explain: + _print_explanations(explain_report(report)) @main.command("health") diff --git a/logsight/reasoning.py b/logsight/reasoning.py new file mode 100644 index 0000000..d6bfdd9 --- /dev/null +++ b/logsight/reasoning.py @@ -0,0 +1,85 @@ +"""Evidence-backed explanations for deterministic LogSight findings. + +This module does not infer a root cause or use a language model. Each explanation +is derived only from the detector's recorded reason, threshold, and observed value. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from logsight.analyzer import AnomalyReport, ErrorRateSpike + +EvidenceStrength = Literal["direct", "statistical"] + + +@dataclass(frozen=True) +class EvidenceExplanation: + """Concise user-facing explanation with machine-readable evidence strings.""" + + category: str + summary: str + evidence_strength: EvidenceStrength + evidence: tuple[str, ...] + + +def explain_report( + report: AnomalyReport, + spikes: list[ErrorRateSpike] | None = None, +) -> list[EvidenceExplanation]: + """Explain detector output without claiming a cause beyond observed evidence.""" + + explanations: list[EvidenceExplanation] = [] + for finding in report.evidence: + entry = finding.entry + if "error_level" in finding.reasons: + explanations.append( + EvidenceExplanation( + category="error-level", + summary=( + f"Entry was flagged because its parsed level is {entry.level.value}; " + "no root cause is inferred." + ), + evidence_strength="direct", + evidence=( + f"level={entry.level.value}", + f"message_length={len(entry.message)}", + ), + ) + ) + if "message_length_zscore" in finding.reasons and finding.message_length_zscore is not None: + explanations.append( + EvidenceExplanation( + category="message-length-outlier", + summary=( + "Entry was flagged because its message length exceeded the configured " + "statistical threshold; no root cause is inferred." + ), + evidence_strength="statistical", + evidence=( + f"zscore={finding.message_length_zscore:.3f}", + f"threshold={report.zscore_threshold:.3f}", + f"message_length={len(entry.message)}", + ), + ) + ) + + for spike in spikes or []: + explanations.append( + EvidenceExplanation( + category="error-rate-spike", + summary=( + f"Entries {spike.start}-{spike.end} crossed the configured error-rate " + "threshold; no root cause is inferred." + ), + evidence_strength="statistical", + evidence=( + f"errors={spike.error_count}", + f"entries={spike.total}", + f"error_rate={spike.error_rate:.3f}", + f"threshold={spike.threshold:.3f}", + ), + ) + ) + return explanations diff --git a/tests/test_analyzer.py b/tests/test_analyzer.py index aa7ecae..4a845fa 100644 --- a/tests/test_analyzer.py +++ b/tests/test_analyzer.py @@ -9,6 +9,7 @@ compute_stats, detect_anomalies, error_rate_spike, + error_rate_spike_details, ) from logsight.parser import LogEntry, LogLevel, parse_line @@ -127,3 +128,25 @@ def test_fewer_than_window(self): entries = [_entry("ERROR", "fail")] * 50 spikes = error_rate_spike(entries, window_size=100, spike_threshold=0.25) assert spikes == [] + + +class TestEvidenceDetails: + def test_records_direct_error_reason(self): + report = detect_anomalies([_entry("ERROR", "database connection failed")]) + assert report.evidence[0].reasons == ("error_level",) + assert report.evidence[0].message_length_zscore is None + + def test_records_zscore_reason(self): + normal = [_entry("INFO", "normal log line")] * 50 + outlier = _entry("INFO", "x" * 500) + report = detect_anomalies(normal + [outlier], zscore_threshold=2.0, flag_errors=False) + assert report.evidence[0].reasons == ("message_length_zscore",) + assert report.evidence[0].message_length_zscore is not None + + def test_spike_details_include_observed_rate(self): + entries = [_entry("ERROR", "fail")] * 2 + [_entry("INFO", "ok")] * 2 + spikes = error_rate_spike_details(entries, window_size=4, spike_threshold=0.5) + assert spikes[0].start == 0 + assert spikes[0].end == 3 + assert spikes[0].error_count == 2 + assert spikes[0].error_rate == 0.5 diff --git a/tests/test_cli.py b/tests/test_cli.py index 4f16564..f37225e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -60,3 +60,12 @@ def test_empty_stdin(self): result = CliRunner().invoke(main, ["stdin"], input="\n") assert result.exit_code == 0 assert "No log entries" in result.output + + def test_explain_uses_detector_evidence(self, tmp_path): + log_file = tmp_path / "events.log" + log_file.write_text("ERROR database connection failed\n", encoding="utf-8") + result = CliRunner().invoke(main, ["analyze", str(log_file), "--explain"]) + assert result.exit_code == 0 + assert "Evidence-backed findings" in result.output + assert "level=ERROR" in result.output + assert "no root cause is inferred" in " ".join(result.output.split()) diff --git a/tests/test_reasoning.py b/tests/test_reasoning.py new file mode 100644 index 0000000..d1e6990 --- /dev/null +++ b/tests/test_reasoning.py @@ -0,0 +1,50 @@ +"""Tests for evidence-backed LogSight explanations.""" + +from __future__ import annotations + +from logsight.analyzer import detect_anomalies, error_rate_spike_details +from logsight.parser import parse_line +from logsight.reasoning import explain_report + + +def _entry(level: str, message: str): + return parse_line(f"{level} {message}") + + +def test_explains_direct_error_evidence_without_causal_claim(): + report = detect_anomalies([_entry("ERROR", "database connection failed")]) + + explanations = explain_report(report) + + assert len(explanations) == 1 + assert explanations[0].category == "error-level" + assert explanations[0].evidence_strength == "direct" + assert "level=ERROR" in explanations[0].evidence + assert "no root cause is inferred" in explanations[0].summary + + +def test_explains_statistical_outlier_and_error_rate_spike(): + entries = [_entry("INFO", "normal log line")] * 50 + [ + _entry("ERROR", "x" * 500), + _entry("ERROR", "failed"), + _entry("INFO", "ok"), + _entry("INFO", "ok"), + ] + report = detect_anomalies(entries, zscore_threshold=2.0) + spikes = error_rate_spike_details(entries[-4:], window_size=4, spike_threshold=0.5) + + explanations = explain_report(report, spikes) + + categories = {explanation.category for explanation in explanations} + assert "error-level" in categories + assert "message-length-outlier" in categories + assert "error-rate-spike" in categories + spike = next(item for item in explanations if item.category == "error-rate-spike") + assert "errors=2" in spike.evidence + assert "entries=4" in spike.evidence + + +def test_returns_no_explanations_when_detector_has_no_findings(): + report = detect_anomalies([_entry("INFO", "healthy")]) + + assert explain_report(report) == []