log-detective

Log Detective — turn a wall of log lines into the three things you needed.

summarise   level counts, time span, busiest minute, top error signatures
signatures  cluster near-identical errors by shape, not text
timeline    events per bucket, so a spike has a timestamp
grep        filtered lines with surrounding context

No network, no credentials, no parsing config. Handles plain text, and reads JSON-lines logs structurally when it detects them.

WHY IT CLUSTERS BY SIGNATURE INSTEAD OF COUNTING LINES

Ten thousand errors are usually four errors. Raw counts hide that: every line differs by a request id, a timestamp, a port number, so nothing groups and you scroll. Normalising the variable parts — numbers, hex ids, UUIDs, paths, quoted strings — collapses them into a handful of shapes, and the shape is the bug.

WHY 'BUSIEST MINUTE' IS ITS OWN NUMBER

An incident is a spike, and a spike is invisible in a total. Knowing that 80% of the day's errors landed inside one minute changes what you go and look at.

WHY IT NEVER PHONES HOME

Logs are the single most PII-dense artifact most systems produce. Anything that uploads them to be "analysed" is a data-egress decision disguised as a feature. This reads local files and returns local results.

Parameters

The typed contract this capability answers to (JSON Schema — the deterministic layer):

{
  "type": "object",
  "properties": {
    "action": {
      "type": "string",
      "enum": [
        "summarise",
        "signatures",
        "timeline",
        "grep"
      ],
      "description": "What to do."
    },
    "path": {
      "type": "string",
      "description": "Path to the log file."
    },
    "pattern": {
      "type": "string",
      "description": "For grep: a regular expression."
    },
    "context": {
      "type": "integer",
      "description": "For grep: lines of context. Default 1."
    },
    "top": {
      "type": "integer",
      "description": "How many signatures/buckets. Default 10."
    }
  },
  "required": [
    "action",
    "path"
  ]
}

Deterministic implementation

Run this instead of improvising when the inputs are well-formed:

def perform(self, **kwargs):
        action, path = kwargs.get("action"), kwargs.get("path")
        if not path or not os.path.isfile(path):
            return json.dumps({"status": "error",
                               "message": f"file not found: {path}"}, indent=2)
        top = int(kwargs.get("top") or 10)
        try:
            lines = _lines(path)
        except Exception as e:
            return json.dumps({"status": "error",
                               "message": f"{type(e).__name__}: {e}"}, indent=2)
        if not lines:
            return json.dumps({"status": "ok", "lines": 0,
                               "note": "file is empty"}, indent=2)

        struct = sum(1 for ln in lines[:200] if _structured(ln))
        jsonl = struct > len(lines[:200]) * 0.6

        def level_of(ln):
            if jsonl:
                s = _structured(ln)
                if s and s["level"]:
                    return s["level"]
            m = LEVEL.search(ln)
            return m.group(1).upper() if m else None

        def body_of(ln):
            if jsonl:
                s = _structured(ln)
                if s and s["message"]:
                    return s["message"]
            return ln

        try:
            if action == "summarise":
                levels = Counter(level_of(l) or "UNLABELLED" for l in lines)
                stamps = [m.group(0) for l in lines
                          for m in [TS.search(l)] if m]
                minutes = Counter(s[:16] for s in stamps)
                bad = [l for l in lines
                       if (level_of(l) or "") in ("ERROR", "ERR", "FATAL", "CRITICAL")]
                sigs = Counter(_norm(body_of(l)) for l in bad)
                busiest = minutes.most_common(1)[0] if minutes else None
                return json.dumps({
                    "status": "ok", "format": "jsonl" if jsonl else "text",
                    "lines": len(lines), "levels": dict(levels.most_common()),
                    "first_timestamp": stamps[0] if stamps else None,
                    "last_timestamp": stamps[-1] if stamps else None,
                    "busiest_minute": ({"minute": busiest[0], "events": busiest[1]}
                                       if busiest else None),
                    "error_lines": len(bad),
                    "top_error_signatures": [
                        {"count": c, "signature": s} for s, c in sigs.most_common(top)],
                    "note": "an incident is a spike, and a spike is invisible in a total",
                }, indent=2)

            if action == "signatures":
                sigs, examples = Counter(), {}
                for l in lines:
                    lv = level_of(l) or ""
                    if lv in ("ERROR", "ERR", "FATAL", "CRITICAL", "WARN", "WARNING"):
                        s = _norm(body_of(l))
                        sigs[s] += 1
                        examples.setdefault(s, l[:200])
                total = sum(sigs.values())
                out = [{"count": c, "share": f"{100.0*c/max(1,total):.0f}%",
                        "signature": s, "example": examples[s]}
                       for s, c in sigs.most_common(top)]
                return json.dumps({
                    "status": "ok", "lines": len(lines),
                    "problem_lines": total, "distinct_signatures": len(sigs),
                    "signatures": out,
                    "note": "ten thousand errors are usually four errors — "
                            "variable parts are normalised so the shape groups",
                }, indent=2)

            if action == "timeline":
                buckets = Counter()
                for l in lines:
                    m = TS.search(l)
                    if m:
                        buckets[m.group(0)[:16]] += 1
                ordered = sorted(buckets.items())
                peak = max(buckets.values()) if buckets else 0
                return json.dumps({
                    "status": "ok", "buckets": len(ordered),
                    "peak_events": peak,
                    "timeline": [{"bucket": b, "events": c,
                                  "bar": "#" * max(1, round(20 * c / max(1, peak)))}
                                 for b, c in ordered[:120]],
                }, indent=2)

            if action == "grep":
                pat = kwargs.get("pattern")
                if not pat:
                    return json.dumps({"status": "error",
                                       "message": "pattern is required for grep"}, indent=2)
                try:
                    rx = re.compile(pat, re.I)
                except re.error as e:
                    return json.dumps({"status": "error",
                                       "message": f"bad regex: {e}"}, indent=2)
                ctx = int(kwargs.get("context") or 1)
                hits = []
                for i, l in enumerate(lines):
                    if rx.search(l):
                        hits.append({"line_no": i + 1,
                                     "before": lines[max(0, i - ctx):i],
                                     "match": l[:300],
                                     "after": lines[i + 1:i + 1 + ctx]})
                    if len(hits) >= 200:
                        break
                return json.dumps({"status": "ok", "lines": len(lines),
                                   "matches": len(hits), "hits": hits[:100]}, indent=2)

            return json.dumps({"status": "error",
                               "message": f"unknown action {action!r}",
                               "valid": ["summarise", "signatures", "timeline", "grep"]}, indent=2)
        except Exception as e:
            return json.dumps({"status": "error",
                               "message": f"{type(e).__name__}: {e}"}, indent=2)