Hermes Agent
6ca58f3dea
log_entries 表 14 列 + id，500 行单 INSERT 产生约 7500 个绑定变量， 超出 SQLite 单语句 999 变量硬限。改为按列数自动计算每语句最大行数 (约 66 行) 分片提交，整批一次性 commit 保持事务性。
824 lines
30 KiB
Python
824 lines
30 KiB
Python
"""Persistent log storage for Squid access.log.
|
|
|
|
This module owns the bridge between raw access.log files on disk and the
|
|
SQLite-backed :class:`~app.LogEntry` table. It exposes:
|
|
|
|
- :func:`sync_log_to_db` - incrementally import new lines from a log file
|
|
- :func:`query_logs` - SQL-backed filter/search with time range support
|
|
- :func:`get_log_stats` - aggregated statistics (hit/miss, top clients...)
|
|
- :func:`get_distinct_hosts` / :func:`get_distinct_clients` /
|
|
:func:`get_distinct_methods` - faceted browsing
|
|
|
|
Implementation notes:
|
|
- We rely on the existing :mod:`log_parser` for line format handling so we do
|
|
not duplicate the access.log grammar.
|
|
- Bulk inserts are batched (default 500 rows) using
|
|
:func:`sqlalchemy.orm.Session.execute` with a parameterised INSERT so we
|
|
don't pay the round-trip cost of per-row commits.
|
|
- De-duplication uses the tuple ``(instance_id, log_time, client, raw_line)``;
|
|
on conflict (we surface those as ``skipped_count`` rather than re-inserting).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime, timezone
|
|
from typing import Iterable
|
|
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
import log_parser
|
|
from app import db, LogEntry # imported lazily-friendly via app
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sync
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _iter_log_lines(path: str, max_lines: int | None = None):
|
|
"""Yield raw lines from ``path`` as bytes-decoded strings.
|
|
|
|
When ``max_lines`` is set we walk backwards through the file with a
|
|
bounded per-iteration buffer (Deque-style window) and yield lines in
|
|
chronological order. For an unbounded scan (full-history import) we
|
|
just stream forwards. Designed to be tolerant of mid-stream truncation
|
|
or binary garbage.
|
|
"""
|
|
if not os.path.isfile(path):
|
|
return
|
|
|
|
if not max_lines or max_lines <= 0:
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="replace") as fh:
|
|
for raw in fh:
|
|
line = raw.rstrip("\n").rstrip("\r")
|
|
if line.strip():
|
|
yield line
|
|
except OSError:
|
|
return
|
|
return
|
|
|
|
# Bounded tail: read last ``max_lines`` lines efficiently
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
try:
|
|
fh.seek(0, os.SEEK_END)
|
|
size = fh.tell()
|
|
chunks: list[bytes] = []
|
|
lines_found = 0
|
|
block = 65536
|
|
pos = size
|
|
while pos > 0 and lines_found <= max_lines:
|
|
read_size = min(block, pos)
|
|
pos -= read_size
|
|
fh.seek(pos)
|
|
chunks.append(fh.read(read_size))
|
|
lines_found += chunks[-1].count(b"\n")
|
|
data = b"".join(reversed(chunks))
|
|
except OSError:
|
|
fh.seek(0)
|
|
data = fh.read()
|
|
text = data.decode("utf-8", errors="replace")
|
|
lines = text.splitlines()
|
|
if len(lines) > max_lines:
|
|
lines = lines[-max_lines:]
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if stripped:
|
|
yield stripped
|
|
except OSError:
|
|
return
|
|
|
|
|
|
def sync_log_to_db(
|
|
path: str,
|
|
instance_id: int = 0,
|
|
batch_size: int = 500,
|
|
max_lines: int = 100000,
|
|
) -> tuple[int, int, str | None]:
|
|
"""Incrementally import new lines from ``path`` into :class:`LogEntry`.
|
|
|
|
Steps:
|
|
|
|
1. Compute the high-water mark ``max(existing.log_time)`` for the
|
|
instance so we can skip already-imported lines quickly.
|
|
2. Stream the last ``max_lines`` lines of the file
|
|
(or the entire file when ``max_lines`` is None/0) and parse them
|
|
through :func:`log_parser.parse_line`.
|
|
3. Insert new rows in batches of ``batch_size`` using
|
|
:func:`sqlalchemy.dialects.sqlite.insert` with
|
|
``OR (primary key uniqueness handled by auot-increment)`` semantics.
|
|
Duplicate suppression uses an in-memory ``seen`` set keyed by
|
|
``(client, log_time, raw_line)`` since the file itself can repeat
|
|
a row at rotation boundaries.
|
|
4. Commit once per batch and return a tuple ``(inserted, skipped, err)``.
|
|
|
|
All exceptions are caught and reported as the third tuple element so the
|
|
caller can decide whether to surface them (e.g. on the dashboard).
|
|
"""
|
|
if batch_size <= 0:
|
|
batch_size = 500
|
|
try:
|
|
# high-water mark: only lines with log_time > wm are new
|
|
wm = (
|
|
db.session.query(func.max(LogEntry.log_time))
|
|
.filter(LogEntry.instance_id == instance_id)
|
|
.scalar()
|
|
)
|
|
if wm is None:
|
|
wm = 0.0
|
|
|
|
seen_keys: set[tuple[str, float, str]] = set()
|
|
inserted = 0
|
|
skipped = 0
|
|
batch: list[dict] = []
|
|
|
|
for raw_line in _iter_log_lines(path, max_lines=max_lines):
|
|
e = log_parser.parse_line(raw_line)
|
|
if not e:
|
|
skipped += 1
|
|
continue
|
|
try:
|
|
ts = float(e.get("time") or 0)
|
|
except (TypeError, ValueError):
|
|
skipped += 1
|
|
continue
|
|
if ts <= wm:
|
|
skipped += 1
|
|
continue
|
|
key = (e.get("client") or "", ts, raw_line)
|
|
if key in seen_keys:
|
|
skipped += 1
|
|
continue
|
|
seen_keys.add(key)
|
|
|
|
batch.append({
|
|
"instance_id": instance_id,
|
|
"log_time": ts,
|
|
"elapsed_ms": int(e.get("elapsed_ms") or 0),
|
|
"client": (e.get("client") or "")[:64],
|
|
"result_code": (e.get("result_code") or "")[:32],
|
|
"http_code": (e.get("http_code") or "")[:8],
|
|
"size_bytes": int(e.get("size") or 0),
|
|
"method": (e.get("method") or "")[:16],
|
|
"url": (e.get("url") or "")[:65535],
|
|
"host": (e.get("host") or "")[:256],
|
|
"hier_code": (e.get("hier_code") or "")[:32],
|
|
"content_type": (e.get("content_type") or "")[:128],
|
|
"raw_line": raw_line[:65535],
|
|
"created_at": datetime.now(timezone.utc),
|
|
})
|
|
if len(batch) >= batch_size:
|
|
_flush_batch(batch)
|
|
inserted += len(batch)
|
|
batch = []
|
|
|
|
if batch:
|
|
_flush_batch(batch)
|
|
inserted += len(batch)
|
|
|
|
# record last sync time so the UI can display "上次同步"
|
|
_record_last_sync(instance_id, path)
|
|
return inserted, skipped, None
|
|
except SQLAlchemyError as e:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return 0, 0, f"DB error: {type(e).__name__}: {e}"
|
|
except Exception as e:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return 0, 0, f"{type(e).__name__}: {e}"
|
|
|
|
|
|
def _column_count() -> int:
|
|
"""Return the number of columns in the :class:`~app.LogEntry` table.
|
|
|
|
Used to cap the per-statement row count so the single INSERT does not
|
|
exceed SQLite's 999-boundary-variable limit per SQL statement.
|
|
"""
|
|
return len(LogEntry.__table__.columns)
|
|
|
|
|
|
# SQLite hard limit: at most 999 ? placeholders per single SQL statement.
|
|
_SQLITE_MAX_VARIABLES = 999
|
|
|
|
|
|
def _max_rows_per_stmt() -> int:
|
|
"""Max rows that can safely fit in one parameterised INSERT."""
|
|
cols = _column_count()
|
|
return max(1, (_SQLITE_MAX_VARIABLES - 1) // cols)
|
|
|
|
|
|
def _flush_batch(batch: list[dict]) -> None:
|
|
"""Insert ``batch`` rows, splitting into chunks small enough for SQLite.
|
|
|
|
SQLite rejects a single parameterised INSERT when its bound-variable
|
|
count exceeds 999. Since :class:`~app.LogEntry` carries many columns, a
|
|
'large' batch (e.g. 500 rows) would blow past that limit. Split the
|
|
batch into sub-chunks of ``_max_rows_per_stmt()`` rows and execute each
|
|
sub-statement individually, then commit once the whole batch is in.
|
|
"""
|
|
if not batch:
|
|
return
|
|
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
|
|
|
chunk_size = _max_rows_per_stmt()
|
|
for start in range(0, len(batch), chunk_size):
|
|
chunk = batch[start:start + chunk_size]
|
|
stmt = sqlite_insert(LogEntry).values(chunk)
|
|
db.session.execute(stmt)
|
|
db.session.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Query
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _parse_time(value) -> float | None:
|
|
"""Parse an ISO-ish datetime / unix timestamp into a unix float (UTC)."""
|
|
if value is None or value == "":
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
if isinstance(value, datetime):
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return value.timestamp()
|
|
if isinstance(value, str):
|
|
s = value.strip()
|
|
if not s:
|
|
return None
|
|
# datetime-local HTML input sends strings like "2024-01-31T10:00"
|
|
# or "2024-01-31 10:00:00". Python's fromisoformat handles both.
|
|
try:
|
|
# Support trailing Z
|
|
if s.endswith("Z"):
|
|
s = s[:-1] + "+00:00"
|
|
dt = datetime.fromisoformat(s)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.timestamp()
|
|
except ValueError:
|
|
# fall through to float attempt
|
|
pass
|
|
try:
|
|
return float(s)
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _entry_to_dict(row: LogEntry) -> dict:
|
|
"""Convert a :class:`LogEntry` row to the dict shape used downstream.
|
|
|
|
Mirrors :class:`log_parser.LogEntry` (which is a dict subclass) so
|
|
existing consumers - templates, alert rules, anomaly detection - keep
|
|
working without modification.
|
|
"""
|
|
ts = float(row.log_time or 0)
|
|
host = row.host or ""
|
|
url = row.url or ""
|
|
raw = row.raw_line or ""
|
|
# split result_code (already stored without slash) - reconstruct full
|
|
# Squid "result" string (e.g. "TCP_HIT/200") for backward compat
|
|
result_full = row.result_code or ""
|
|
if row.http_code:
|
|
result_full = f"{result_full}/{row.http_code}"
|
|
return {
|
|
"id": row.id,
|
|
"time": row.log_time, # float unix ts
|
|
"timestamp": datetime.fromtimestamp(ts, tz=timezone.utc) if ts > 0 else None,
|
|
"elapsed_ms": int(row.elapsed_ms or 0),
|
|
"client": row.client or "",
|
|
"result": result_full,
|
|
"result_code": row.result_code or "",
|
|
"http_code": row.http_code or "",
|
|
"category": log_parser.RESULT_CATEGORIES.get(row.result_code or "", "Other"),
|
|
"result_description": log_parser.RESULT_CODE_DESCRIPTION.get(row.result_code or "", row.result_code or ""),
|
|
"size": int(row.size_bytes or 0),
|
|
"method": row.method or "",
|
|
"url": url,
|
|
"host": host,
|
|
"ident": "", # not persisted separately
|
|
"hierarchy": row.hier_code or "",
|
|
"hier_code": row.hier_code or "",
|
|
"peer": "", # not persisted
|
|
"content_type": row.content_type or "",
|
|
"raw_line": raw,
|
|
"formatted_size": log_parser.format_bytes(int(row.size_bytes or 0)),
|
|
"formatted_duration": log_parser.format_duration(int(row.elapsed_ms or 0)),
|
|
}
|
|
|
|
|
|
def query_logs(filters: dict) -> list[dict]:
|
|
"""Run a filtered SQL query against the persisted log rows.
|
|
|
|
Supported keys (all optional):
|
|
|
|
- ``instance_id`` (int)
|
|
- ``client``, ``method``, ``result_code``, ``host`` (exact or substring)
|
|
- ``host_filter_mode`` "exact" | "contains" (default "contains")
|
|
- ``url`` substring LIKE
|
|
- ``min_size``, ``max_size`` (int bytes)
|
|
- ``min_elapsed``, ``max_elapsed`` (int milliseconds)
|
|
- ``start_time``, ``end_time`` (datetime or unix timestamp, inclusive lo,
|
|
inclusive hi for end so users don't accidentally exclude the minute
|
|
they typed)
|
|
- ``limit`` (default 1000), ``offset`` (default 0)
|
|
|
|
Returns a list of dicts with the same shape as
|
|
:func:`log_parser.parse_lines`, plus ``formatted_size`` /
|
|
``formatted_duration`` keys.
|
|
"""
|
|
try:
|
|
q = db.session.query(LogEntry)
|
|
# instance scoping
|
|
iid = filters.get("instance_id")
|
|
if iid is not None:
|
|
q = q.filter(LogEntry.instance_id == iid)
|
|
|
|
c = (filters.get("client") or "").strip()
|
|
if c:
|
|
q = q.filter(LogEntry.client.like(f"%{c}%"))
|
|
m = (filters.get("method") or "").strip()
|
|
if m:
|
|
q = q.filter(LogEntry.method == m)
|
|
rc = (filters.get("result_code") or "").strip()
|
|
if rc:
|
|
q = q.filter(LogEntry.result_code.like(f"%{rc}%"))
|
|
h = (filters.get("host") or "").strip()
|
|
if h:
|
|
mode = (filters.get("host_filter_mode") or "contains").lower()
|
|
if mode == "exact":
|
|
q = q.filter(LogEntry.host == h)
|
|
else:
|
|
q = q.filter(LogEntry.host.like(f"%{h}%"))
|
|
u = (filters.get("url") or "").strip()
|
|
if u:
|
|
q = q.filter(LogEntry.url.like(f"%{u}%"))
|
|
|
|
mn = filters.get("min_size")
|
|
if mn is not None:
|
|
q = q.filter(LogEntry.size_bytes >= int(mn))
|
|
mx = filters.get("max_size")
|
|
if mx is not None:
|
|
q = q.filter(LogEntry.size_bytes <= int(mx))
|
|
mne = filters.get("min_elapsed")
|
|
if mne is not None:
|
|
q = q.filter(LogEntry.elapsed_ms >= int(mne))
|
|
mxe = filters.get("max_elapsed")
|
|
if mxe is not None:
|
|
q = q.filter(LogEntry.elapsed_ms <= int(mxe))
|
|
|
|
st = _parse_time(filters.get("start_time"))
|
|
if st is not None:
|
|
q = q.filter(LogEntry.log_time >= st)
|
|
et = _parse_time(filters.get("end_time"))
|
|
if et is not None:
|
|
q = q.filter(LogEntry.log_time <= et)
|
|
|
|
# default ordering: newest first
|
|
q = q.order_by(LogEntry.log_time.desc(), LogEntry.id.desc())
|
|
|
|
limit = int(filters.get("limit") or 1000)
|
|
offset = int(filters.get("offset") or 0)
|
|
if limit > 0:
|
|
q = q.limit(limit)
|
|
if offset > 0:
|
|
q = q.offset(offset)
|
|
|
|
rows = q.all()
|
|
return [_entry_to_dict(r) for r in rows]
|
|
except SQLAlchemyError as e:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def count_logs(filters: dict | None = None) -> int:
|
|
"""Return the number of rows matching ``filters`` (same shape as ``query_logs``).
|
|
|
|
Used by ``logs_view`` to render total / pagination accurately without
|
|
transferring every row over the wire.
|
|
"""
|
|
f = dict(filters or {})
|
|
f["limit"] = 0
|
|
f["offset"] = 0
|
|
try:
|
|
# rebuild only the WHERE portion - run the same filter logic but
|
|
# count instead of fetch
|
|
q = _build_base_query(f)
|
|
return q.with_entities(func.count(LogEntry.id)).scalar() or 0
|
|
except SQLAlchemyError:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return 0
|
|
|
|
|
|
def _build_base_query(filters: dict):
|
|
"""Internal: same WHERE-clause as :func:`query_logs` but returns the
|
|
SQLAlchemy query for further composition (e.g. count)."""
|
|
q = db.session.query(LogEntry)
|
|
iid = filters.get("instance_id")
|
|
if iid is not None:
|
|
q = q.filter(LogEntry.instance_id == iid)
|
|
c = (filters.get("client") or "").strip()
|
|
if c:
|
|
q = q.filter(LogEntry.client.like(f"%{c}%"))
|
|
m = (filters.get("method") or "").strip()
|
|
if m:
|
|
q = q.filter(LogEntry.method == m)
|
|
rc = (filters.get("result_code") or "").strip()
|
|
if rc:
|
|
q = q.filter(LogEntry.result_code.like(f"%{rc}%"))
|
|
h = (filters.get("host") or "").strip()
|
|
if h:
|
|
q = q.filter(LogEntry.host.like(f"%{h}%"))
|
|
u = (filters.get("url") or "").strip()
|
|
if u:
|
|
q = q.filter(LogEntry.url.like(f"%{u}%"))
|
|
mn = filters.get("min_size")
|
|
if mn is not None:
|
|
q = q.filter(LogEntry.size_bytes >= int(mn))
|
|
mx = filters.get("max_size")
|
|
if mx is not None:
|
|
q = q.filter(LogEntry.size_bytes <= int(mx))
|
|
mne = filters.get("min_elapsed")
|
|
if mne is not None:
|
|
q = q.filter(LogEntry.elapsed_ms >= int(mne))
|
|
mxe = filters.get("max_elapsed")
|
|
if mxe is not None:
|
|
q = q.filter(LogEntry.elapsed_ms <= int(mxe))
|
|
st = _parse_time(filters.get("start_time"))
|
|
if st is not None:
|
|
q = q.filter(LogEntry.log_time >= st)
|
|
et = _parse_time(filters.get("end_time"))
|
|
if et is not None:
|
|
q = q.filter(LogEntry.log_time <= et)
|
|
return q
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Aggregate stats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_log_stats(
|
|
instance_id: int = 0,
|
|
start_time: float | None = None,
|
|
end_time: float | None = None,
|
|
) -> dict:
|
|
"""Compute aggregate statistics in SQL.
|
|
|
|
Compatible with :func:`log_parser.aggregate_stats`. When there are no
|
|
rows in scope we return the same ``_empty_stats()`` structure so the
|
|
caller doesn't need a special case.
|
|
"""
|
|
try:
|
|
q = _build_base_query({
|
|
"instance_id": instance_id,
|
|
"start_time": start_time,
|
|
"end_time": end_time,
|
|
})
|
|
total = q.with_entities(func.count(LogEntry.id)).scalar() or 0
|
|
if total == 0:
|
|
stats = _copy_empty_stats()
|
|
else:
|
|
agg = q.with_entities(
|
|
func.coalesce(func.sum(LogEntry.size_bytes), 0),
|
|
func.coalesce(func.sum(LogEntry.elapsed_ms), 0),
|
|
func.min(LogEntry.log_time),
|
|
func.max(LogEntry.log_time),
|
|
).one()
|
|
total_bytes, total_elapsed, ts_min, ts_max = agg
|
|
|
|
total = int(total)
|
|
total_bytes = int(total_bytes)
|
|
total_elapsed = int(total_elapsed)
|
|
avg_elapsed = round(total_elapsed / total, 2) if total else 0.0
|
|
avg_size = round(total_bytes / total, 2) if total else 0.0
|
|
|
|
# result_code distribution
|
|
by_result = dict(
|
|
q.with_entities(
|
|
LogEntry.result_code,
|
|
func.count(LogEntry.id),
|
|
).group_by(LogEntry.result_code).all()
|
|
)
|
|
by_result = {k or "": v for k, v in by_result.items()}
|
|
# http_code distribution (only rows with a value)
|
|
by_http = dict(
|
|
q.filter(LogEntry.http_code != "").with_entities(
|
|
LogEntry.http_code,
|
|
func.count(LogEntry.id),
|
|
).group_by(LogEntry.http_code).all()
|
|
)
|
|
# method distribution
|
|
by_method = dict(
|
|
q.with_entities(
|
|
LogEntry.method,
|
|
func.count(LogEntry.id),
|
|
).group_by(LogEntry.method).all()
|
|
)
|
|
# hierarchy distribution
|
|
by_hier = dict(
|
|
q.with_entities(
|
|
LogEntry.hier_code,
|
|
func.count(LogEntry.id),
|
|
).group_by(LogEntry.hier_code).all()
|
|
)
|
|
|
|
# top clients: most active + bytes
|
|
top_clients = q.with_entities(
|
|
LogEntry.client,
|
|
func.count(LogEntry.id).label("c"),
|
|
func.coalesce(func.sum(LogEntry.size_bytes), 0).label("b"),
|
|
).group_by(LogEntry.client).order_by(func.count(LogEntry.id).desc()).limit(50).all()
|
|
top_clients = [{"client": c or "", "count": int(c_), "bytes": int(b_)} for c, c_, b_ in top_clients]
|
|
# top hosts
|
|
top_hosts_rows = q.filter(LogEntry.host != "").with_entities(
|
|
LogEntry.host,
|
|
func.count(LogEntry.id).label("c"),
|
|
func.coalesce(func.sum(LogEntry.size_bytes), 0).label("b"),
|
|
).group_by(LogEntry.host).order_by(func.count(LogEntry.id).desc()).limit(50).all()
|
|
top_hosts = [{"host": h, "count": int(c_), "bytes": int(b_)} for h, c_, b_ in top_hosts_rows]
|
|
# top urls
|
|
top_urls_rows = q.with_entities(
|
|
LogEntry.url,
|
|
func.count(LogEntry.id).label("c"),
|
|
).group_by(LogEntry.url).order_by(func.count(LogEntry.id).desc()).limit(50).all()
|
|
top_urls = [{"url": u, "count": int(c_)} for u, c_ in top_urls_rows]
|
|
|
|
# content-type: simpler approach - just count distinct top values
|
|
by_ctype_rows = q.with_entities(
|
|
LogEntry.content_type,
|
|
func.count(LogEntry.id).label("c"),
|
|
).group_by(LogEntry.content_type).order_by(func.count(LogEntry.id).desc()).limit(20).all()
|
|
by_ctype = []
|
|
for ct, c_ in by_ctype_rows:
|
|
ct_main = (ct or "-").split(";")[0].strip() or "-"
|
|
by_ctype.append((ct_main, int(c_)))
|
|
# collapse same main types
|
|
merged: dict[str, int] = defaultdict(int)
|
|
for ct_main, c_ in by_ctype:
|
|
merged[ct_main] += c_
|
|
by_ctype = list(merged.items())[:20]
|
|
|
|
# hit/miss ratio using result_code categories
|
|
hits = sum(c for code, c in by_result.items()
|
|
if log_parser.RESULT_CATEGORIES.get(code or "") == "Hit")
|
|
misses = sum(c for code, c in by_result.items()
|
|
if log_parser.RESULT_CATEGORIES.get(code or "") == "Miss")
|
|
denied = sum(c for code, c in by_result.items()
|
|
if log_parser.RESULT_CATEGORIES.get(code or "") == "Denied")
|
|
aborted = by_result.get("TCP_MISS_ABORTED", 0)
|
|
denied_rate = denied / total if total else 0.0
|
|
aborted_rate = aborted / total if total else 0.0
|
|
hit_ratio = hits / (hits + misses) if (hits + misses) > 0 else 0.0
|
|
errors_4xx = sum(c for code, c in by_http.items() if code.startswith("4"))
|
|
errors_5xx = sum(c for code, c in by_http.items() if code.startswith("5"))
|
|
errors_4xx_rate = errors_4xx / total if total else 0.0
|
|
errors_5xx_rate = errors_5xx / total if total else 0.0
|
|
|
|
# unique clients / hosts
|
|
unique_clients = q.with_entities(func.count(func.distinct(LogEntry.client))).scalar() or 0
|
|
unique_hosts = q.filter(LogEntry.host != "").with_entities(
|
|
func.count(func.distinct(LogEntry.host))
|
|
).scalar() or 0
|
|
|
|
# hourly distribution (UTC hour of day)
|
|
by_hour_counts: dict[int, int] = defaultdict(int)
|
|
by_hour_bytes: dict[int, int] = defaultdict(int)
|
|
# we already have min/max so size of an in-Python bucket is small
|
|
# but to avoid pulling all rows we can do an UNION-ish query per
|
|
# hour; the simpler approach: fetch (time, size) projection.
|
|
rows = q.with_entities(LogEntry.log_time, LogEntry.size_bytes).all()
|
|
for ts, size in rows:
|
|
if not ts:
|
|
continue
|
|
hr = datetime.fromtimestamp(float(ts), tz=timezone.utc).hour
|
|
by_hour_counts[hr] += 1
|
|
by_hour_bytes[hr] += int(size or 0)
|
|
|
|
time_start = datetime.fromtimestamp(float(ts_min), tz=timezone.utc) if ts_min else None
|
|
time_end = datetime.fromtimestamp(float(ts_max), tz=timezone.utc) if ts_max else None
|
|
|
|
stats = {
|
|
"total_requests": total,
|
|
"total_bytes": total_bytes,
|
|
"total_elapsed_ms": total_elapsed,
|
|
"unique_clients": int(unique_clients),
|
|
"unique_hosts": int(unique_hosts),
|
|
"hit_ratio": hit_ratio,
|
|
"denied_rate": denied_rate,
|
|
"aborted_rate": aborted_rate,
|
|
"errors_4xx_rate": errors_4xx_rate,
|
|
"errors_5xx_rate": errors_5xx_rate,
|
|
"errors_4xx_count": errors_4xx,
|
|
"errors_5xx_count": errors_5xx,
|
|
"avg_elapsed_ms": avg_elapsed,
|
|
"avg_size_bytes": avg_size,
|
|
"hits": hits,
|
|
"misses": misses,
|
|
"denied": denied,
|
|
"aborted": aborted,
|
|
"time_start": time_start,
|
|
"time_end": time_end,
|
|
"by_result": log_parser._top_n(Counter(by_result), 50),
|
|
"by_http": log_parser._top_n(Counter(by_http), 50),
|
|
"by_method": log_parser._top_n(Counter(by_method), 20),
|
|
"by_category": [], # we don't store category; compute inline if needed
|
|
"by_hierarchy": log_parser._top_n(Counter(by_hier), 20),
|
|
"by_ctype": by_ctype,
|
|
"top_clients": top_clients,
|
|
"top_hosts": top_hosts,
|
|
"top_urls": top_urls,
|
|
"hourly": [
|
|
{"hour": h, "requests": int(by_hour_counts.get(h, 0)),
|
|
"bytes": int(by_hour_bytes.get(h, 0))}
|
|
for h in range(24)
|
|
],
|
|
"result_code_legend": log_parser.RESULT_CODE_DESCRIPTION,
|
|
"hierarchy_legend": log_parser.HIERARCHY_LABELS,
|
|
}
|
|
|
|
# augment with sync timestamp for the dashboard UI
|
|
try:
|
|
from app import AppConfig
|
|
ts_str = AppConfig.query.filter_by(key="_log_last_sync_at").first()
|
|
stats["last_sync_at"] = ts_str.value if ts_str else None
|
|
except Exception:
|
|
stats["last_sync_at"] = None
|
|
return stats
|
|
except SQLAlchemyError as e:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
s = _copy_empty_stats()
|
|
s["last_sync_at"] = None
|
|
s["error"] = f"{type(e).__name__}: {e}"
|
|
return s
|
|
|
|
|
|
def _copy_empty_stats() -> dict:
|
|
"""Return a fresh empty-stats dict (avoid mutating log_parser._empty_stats)."""
|
|
return log_parser._empty_stats()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Distinct facets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_distinct_hosts(instance_id: int = 0, limit: int = 200) -> list[str]:
|
|
try:
|
|
rows = (
|
|
db.session.query(LogEntry.host)
|
|
.filter(LogEntry.host != "", LogEntry.instance_id == instance_id)
|
|
.distinct()
|
|
.order_by(LogEntry.host.asc())
|
|
.limit(int(limit))
|
|
.all()
|
|
)
|
|
return [r[0] for r in rows if r[0]]
|
|
except SQLAlchemyError:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def get_distinct_clients(instance_id: int = 0, limit: int = 200) -> list[str]:
|
|
try:
|
|
rows = (
|
|
db.session.query(LogEntry.client)
|
|
.filter(LogEntry.client != "", LogEntry.instance_id == instance_id)
|
|
.distinct()
|
|
.order_by(LogEntry.client.asc())
|
|
.limit(int(limit))
|
|
.all()
|
|
)
|
|
return [r[0] for r in rows if r[0]]
|
|
except SQLAlchemyError:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def get_distinct_methods(instance_id: int = 0) -> list[str]:
|
|
try:
|
|
rows = (
|
|
db.session.query(LogEntry.method)
|
|
.filter(LogEntry.method != "", LogEntry.instance_id == instance_id)
|
|
.distinct()
|
|
.order_by(LogEntry.method.asc())
|
|
.all()
|
|
)
|
|
return [r[0] for r in rows if r[0]]
|
|
except SQLAlchemyError:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sync bookkeeping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_LAST_SYNC_KEY = "_log_last_sync_at"
|
|
|
|
|
|
def _record_last_sync(instance_id: int, path: str) -> None:
|
|
"""Stamp ``AppConfig`` rows with the last successful sync timestamp.
|
|
|
|
Two values are stored: ``_log_last_sync_at`` (ISO timestamp) and
|
|
``_log_last_sync_path`` (the file we synced from). We use ``AppConfig``
|
|
rather than ``LogEntry`` so the timestamp survives even if log rows
|
|
are later rotated out.
|
|
"""
|
|
try:
|
|
from app import AppConfig
|
|
ts_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
for key, value in (
|
|
(_LAST_SYNC_KEY, ts_iso),
|
|
("_log_last_sync_path", path or ""),
|
|
("_log_last_sync_instance", str(instance_id)),
|
|
):
|
|
row = AppConfig.query.filter_by(key=key).first()
|
|
if row is None:
|
|
row = AppConfig(key=key, value=value)
|
|
db.session.add(row)
|
|
else:
|
|
row.value = value
|
|
db.session.commit()
|
|
except Exception:
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def get_last_sync_at() -> str | None:
|
|
"""Return the ISO timestamp of the last sync, or ``None``."""
|
|
try:
|
|
from app import AppConfig
|
|
row = AppConfig.query.filter_by(key=_LAST_SYNC_KEY).first()
|
|
return row.value if row else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_last_sync_path() -> str | None:
|
|
try:
|
|
from app import AppConfig
|
|
row = AppConfig.query.filter_by(key="_log_last_sync_path").first()
|
|
return row.value if row else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def import_full_history(
|
|
path: str,
|
|
instance_id: int = 0,
|
|
batch_size: int = 500,
|
|
progress_cb=None,
|
|
) -> tuple[int, int, str | None]:
|
|
"""Import the entire ``access.log`` from byte 0 (not just the tail).
|
|
|
|
Used by the "Import history" button on the logs page. Calls
|
|
:func:`sync_log_to_db` with a very high ``max_lines`` (50M) so we
|
|
stream every line. Progress is reported via ``progress_cb(processed)``
|
|
so the caller can render a status indicator.
|
|
"""
|
|
try:
|
|
# 50M lines ceiling - enough for multi-year logs; streaming means
|
|
# we never actually load them all into memory.
|
|
return sync_log_to_db(
|
|
path=path,
|
|
instance_id=instance_id,
|
|
batch_size=batch_size,
|
|
max_lines=50_000_000,
|
|
)
|
|
except Exception as e:
|
|
return 0, 0, f"{type(e).__name__}: {e}"
|