diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ca650dc..67e3406 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,10 +20,27 @@ jobs: with: python-version: ${{ matrix.python-version }} - - run: pip install -e ".[dev,http]" + - run: pip install -e ".[dev,http,apprise]" - run: ruff check . - run: mypy logquill - run: pytest --cov + + # Memory budgets for the logging hot path (see benchmarks/measure.py). Kept + # out of the matrix above and away from coverage: line tracing inflates + # allocation counts, and the budgets only need one interpreter to gate on. + benchmarks: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v7 + + - uses: actions/setup-python@v7 + with: + python-version: "3.12" + + - run: pip install -e ".[dev]" + + - run: pytest benchmarks diff --git a/CHANGELOG.md b/CHANGELOG.md index 0bfad1f..e754681 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,62 @@ All notable changes to this project are documented in this file. +## Unreleased + +- Added the 1.0 features that hadn't shipped yet, plus hardening: + - `logger.opt(lazy=True)` defers callable `meta` values until a record is + really going to be emitted, so an expensive `DEBUG`/`TRACE` argument costs + nothing when the level filters the call out. A callable that raises leaves + a placeholder in the record instead of crashing the caller. + `logger.opt(depth=N)` adds `meta.caller` (`module`, `function`, `line`, + `file`) naming the code that logged, `N` frames up, so a call made from a + wrapper or decorator reports the wrapper's caller. + - `logquill.disable(name)` / `logquill.enable(name)` switch off a logger and + everything nested under it (most specific rule wins), so a library that + logs through LogQuill can be silent in its host application by default — + `logquill.disable(__name__)` — and the application can turn it back on. + - Two new formatters: `TextFormatter` (a human-readable entry per record, + with tracebacks on the lines below) and `LogfmtFormatter` (single-line + `key=value` output; values are quoted so a record is always one line). + Formatters now live in the `logquill.formatters` package; + `logquill.formatter` still works. A transport's `options` in a config file + can name one: `{"formatter": "text"}`. `logquill tail` now prints + tracebacks the same way `TextFormatter` does. + - `parse(source, pattern, cast=...)` extracts structured fields from a log + file with a regex — including legacy and third-party formats — streaming + line by line. `parse_logfmt()` reads logfmt back, and `TEXT_LOG_PATTERN` + reads `TextFormatter` output. + - `AppriseAlertPlugin` (`pip install logquill[apprise]`) sends alerts + through Apprise, reaching 100+ notification services with the same + deduplication and non-blocking behavior as the other alerting plugins. + - Every `Logger` now flushes and closes its transports at interpreter exit + (an `atexit` hook), so a script that never calls `close()` no longer loses + its last queued records or a batching transport's unsent batch. Opt out + with `Logger(flush_at_exit=False)` or `"flush_at_exit": false` in config. + - `diagnose=True` on any `Logger` method adds each traceback frame's local + variable values. Off by default, with an explicit warning in the docs and + once per process in the log: it can leak sensitive data. Captured values + go through `RedactPlugin` and `PIIRedactPlugin` *before* the traceback is + formatted (via a new optional `Plugin.redact_local` hook), and a plugin + whose hook raises masks the value instead of showing it. + - `HTTPTransport(backend="aiohttp")` sends over one reused keep-alive + connection, giving the `http` extra a purpose. Also, `HTTPTransport` now + bounds its buffer by `max_bytes` as well as `batch_size`, and a failed + send is logged with an actionable message and the batch dropped, rather + than raising into the code that logged. + - Fixed: the async queue's "dropping records" warning could stay silent for + the first minute after a process started, because its rate limiter + compared against a monotonic clock whose zero point is arbitrary. + - Fixed: passing a malformed `exc_info` (for example a forwarded dict that + happens to carry a bad `exc_info` key) raised out of the log call. It is + now ignored with a warning, like any other bad `meta`. + - A memory-budget suite (`pytest benchmarks`, its own CI job) fails the + build if a log call, a level-filtered call, or a burst into a stalled + sink uses materially more memory than it does today. New tests burst + 30,000 records at a stalled transport under each backpressure policy and + assert exactly which records survive, and `hypothesis` coverage now + extends to the formatters and transports. + ## 1.0.0 - 2026-09-05 - First stable release: bumped the `Development Status` classifier from diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 6a23d5f..235adf8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,8 +15,15 @@ pre-commit install ruff check . mypy logquill pytest +pytest benchmarks # memory budgets for the logging hot path ``` +`pytest benchmarks` measures how much memory a log call and a stalled-sink +burst use, and fails if a change pushes either past its budget in +`benchmarks/measure.py`. It's separate from `pytest` because coverage's line +tracing distorts the numbers; CI runs it as its own job. If a change +legitimately needs more memory, raise the budget in the same PR and say why. + ## Pull request strategy - **Branch from `main`**, name branches by intent: `feat/…`, `fix/…`, @@ -31,7 +38,7 @@ pytest 4. `CHANGELOG.md` has an entry under `Unreleased` 5. Nothing in the cross-language contract table silently diverged from `logquill-js` (open a tracking issue there if it changed) -- **CI must be green** (`ruff check`, `mypy logquill`, `pytest`) and **at +- **CI must be green** (`ruff check`, `mypy logquill`, `pytest`, `pytest benchmarks`) and **at least one review approval** is required before merge — enforced by branch protection on `main`. - **Squash-merge** into `main` — keep the squash commit message a clear diff --git a/README.md b/README.md index a5f0d43..f31a810 100644 --- a/README.md +++ b/README.md @@ -23,14 +23,16 @@ for what's landed so far. - **Structured by default** — every call carries a `meta` dict, not just a message string - **Cross-language record shape** — identical JSON shape and level names/weights as [`logquill` on npm](https://www.npmjs.com/package/logquill) - **Pluggable transports** — `ConsoleTransport` (colorized, stderr for errors), `FileTransport` (rotation, optional encryption-at-rest), `HTTPTransport` (batched), `SyslogTransport` (RFC 5424, UDP/TCP), plus SQL/NoSQL/message-queue/cloud-native sinks (see [Transports](#transports)); write your own by subclassing `Transport` -- **Pluggable formatters** — `JSONFormatter` out of the box; implement `format(record) -> str` for your own +- **Pluggable formatters** — `JSONFormatter` (default, machine-readable), `TextFormatter` (human-readable, for terminals), and `LogfmtFormatter` (`key=value`, the Heroku/Go convention); implement `format(record) -> str` for your own — see [Formatters](#formatters) - **Config from file/env** — `load_config(dict)`, `logger_from_file(path)` (JSON/YAML), `logger_from_env()` build a `Logger` from one config shape — see [Config](#config) - **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), `TraceContextPlugin` (cross-service trace correlation), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins)) - **Agentic & harness tracing** — `.child()` loggers, `RunPlugin`, `.thought()/.action()/.observation()/.decision()`, `with agent_log.span(...)`, and framework adapters — `LangChainAdapter` (`pip install logquill[langchain]`), `LangGraphAdapter` (`pip install logquill[langgraph]`, adds checkpoint interrupt/resume events on top), `CrewAIAdapter` (`pip install logquill[crewai]`), `LlamaIndexAdapter` (`pip install logquill[llamaindex]`), and `AutoGenAdapter` (`pip install logquill[autogen]`) — see [Agentic & harness tracing](#agentic--harness-tracing) - **Non-blocking async dispatch** — `Logger(async_dispatch=True)` moves transport writes onto a background thread with a bounded queue and a configurable backpressure policy (`drop_oldest`/`drop_newest`/`block`); `flush()`/`flush_async()` and a `with_lambda`/`with_cloud_function`/`with_azure_function` decorator make serverless shutdown safe — see [Async dispatch & serverless safety](#async-dispatch--serverless-safety) -- **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP +- **Zero required runtime dependencies** — stdlib only; `aiohttp` (`logquill[http]`) is opt-in, for keep-alive HTTP delivery - **Typed throughout** — `mypy --strict` clean on the public API - **Context propagation, exception capture & the stdlib bridge** — `bind_context()` (`contextvars`-based, no manual passing), `exc_info=` on any `Logger` method (formatted traceback into `meta["stack"]`), `LogQuillHandler` (bridges stdlib `logging` into a `Logger`), and `RateLimitPlugin` — see [Context propagation, exception capture & the stdlib bridge](#context-propagation-exception-capture--the-stdlib-bridge) +- **Cheap when idle, precise when it counts** — `logger.opt(lazy=True)` defers expensive `meta` values until a record will really be emitted, `logger.opt(depth=N)` reports the right caller from inside a wrapper, `logquill.disable(__name__)` silences a library's own logs by default, and queued records are flushed automatically at interpreter exit — see [Lazy values, caller depth & disabling a library](#lazy-values-caller-depth--disabling-a-library) +- **Parse any log file** — `parse()` pulls structured fields out of a log file (LogQuill's own or a legacy format) with a regex, streaming line by line — see [Parsing log files](#parsing-log-files) - **CLI** — `logquill tail app.log --level=warn --json -f` for filtering/following a JSONL log file in local dev, no extra install — see [CLI](#cli) ## Install @@ -126,8 +128,9 @@ logger = load_config({ ## Transports Attach transports to a `Logger` to actually write records somewhere. Each -record is dispatched to every attached transport synchronously (non-blocking -dispatch isn't implemented yet): +record is dispatched to every attached transport synchronously by default; +pass `async_dispatch=True` to move that onto a background thread (see +[Async dispatch & serverless safety](#async-dispatch--serverless-safety)): ```python from logquill import ConsoleTransport, FileTransport, HTTPTransport, Logger @@ -160,6 +163,63 @@ logger.info("hello") assert sink.records[0]["message"] == "hello" ``` +`HTTPTransport` sends with stdlib `urllib` by default. `backend="aiohttp"` +(`pip install logquill[http]`) sends over one reused keep-alive connection +instead, which saves a TCP/TLS handshake per batch against an HTTPS collector: + +```python +from logquill import HTTPTransport + +transport = HTTPTransport("https://logs.example.com/ingest", backend="aiohttp", timeout=5.0) +transport.close() # sends anything buffered and closes the connection +``` + +Its buffer is bounded by both `batch_size` records and `max_bytes` of +formatted text, and a failed send is logged (naming the URL) and that batch +dropped — it never raises into the code that logged. + +### Formatters + +A transport renders each record with its `formatter`. `JSONFormatter` is the +default and the right choice for anything a machine reads. Two more ship for +other readers: + +- `TextFormatter` — one human-readable entry per record, with any traceback + printed on the lines below it. For terminals and local development. +- `LogfmtFormatter` — a single `key=value` line, the Heroku/Go convention, + for tools (Loki, Splunk, `grep`) that expect it. Nested `meta` flattens to + dotted keys, and a value with spaces or newlines is quoted, so a record is + always exactly one line. + +```python +import io + +from logquill import ConsoleTransport, LogfmtFormatter, Logger, TextFormatter, parse_logfmt + +text_out, logfmt_out = io.StringIO(), io.StringIO() +logger = Logger( + "app.api", + transports=[ + ConsoleTransport(formatter=TextFormatter(), colorize=False, stdout=text_out), + ConsoleTransport(formatter=LogfmtFormatter(), colorize=False, stdout=logfmt_out), + ], +) + +logger.info("user signed up", user_id=42, http={"status": 201}, note="from the web form") + +assert 'app.api: user signed up {"user_id":42,' in text_out.getvalue() + +fields = parse_logfmt(logfmt_out.getvalue()) +assert fields["level"] == "INFO" +assert fields["user_id"] == "42" +assert fields["http.status"] == "201" +assert fields["note"] == "from the web form" +``` + +In a config file, name the formatter as a string in the transport's +`options`: `{"type": "console", "options": {"formatter": "text"}}` (`"json"`, +`"text"` or `"logfmt"`). To write your own, implement `format(record) -> str`. + ### SQL, NoSQL, message queue, and cloud-native transports Every transport below shares one design: records are **always batched** @@ -491,6 +551,27 @@ Write your own destination by subclassing `AlertingPlugin` and implementing `send_alert(record, occurrences)`; thresholding, deduplication, and the never-block-the-caller behavior are all handled by the base class. +`AppriseAlertPlugin` reaches everything else. It hands the alert to +[Apprise](https://github.com/caronc/apprise) (`pip install logquill[apprise]`), +which speaks to 100+ services — Discord, Telegram, Microsoft Teams, ntfy, +Matrix, SMS gateways — from one URL each, so you don't need a plugin per +service. It has the same background-thread sending, deduplication and +failure handling as the other alerting plugins. Prefer `SlackAlertPlugin` or +`PagerDutyAlertPlugin` for those two, which format richer messages than +Apprise's generic title-and-body allows: + +```python +from logquill import AppriseAlertPlugin, Logger + +logger = Logger( + "app", + plugins=[AppriseAlertPlugin(["discord://webhook_id/webhook_token", "ntfy://my-topic"])], +) +``` + +A URL Apprise doesn't recognize raises `ValueError` right away, at startup, +instead of silently failing on the first real alert. + ## Agentic & harness tracing `.child()` makes a namespaced logger that shares the parent's transports — @@ -741,6 +822,16 @@ closes every transport: logger.close(timeout=5.0) ``` +If a script ends without calling `close()`, LogQuill does it for you: by +default every `Logger` registers an `atexit` hook that drains the queue +(waiting up to 5 seconds) and closes its transports once, so the last records +and a batching transport's unsent batch aren't lost. Pass +`flush_at_exit=False` (or `"flush_at_exit": false` in a config file) if you'd +rather manage shutdown yourself. `atexit` only runs on a normal exit — end of +script, `sys.exit()`, an unhandled exception — not when the process is killed +outright (`SIGKILL`, `os._exit()`, or `SIGTERM` with no handler), which is +why the Kubernetes note below still applies. + For code that keeps running afterward (a request handler, a serverless invocation), use `logger.flush()` instead — it drains the queue and flushes each transport's own internal buffer (see `BatchingTransport`) *without* @@ -853,6 +944,62 @@ assert record["meta"]["order_id"] == 42 assert "ZeroDivisionError" in record["meta"]["stack"] ``` +### Local variables in tracebacks: `diagnose` + +`diagnose=True` adds each frame's local variable values under its source line, +which turns "it failed in `charge()`" into "it failed because `amount` was +`0`": + +```python +from logquill import Logger + +logger = Logger("app") + + +def charge(amount): + fee = 2.5 + return fee / amount + + +try: + charge(0) +except ZeroDivisionError: + record = logger.error("payment failed", diagnose=True) # implies exc_info=True + +assert "amount = 0" in record["meta"]["stack"] +assert "fee = 2.5" in record["meta"]["stack"] +``` + +**It's off by default, and it can leak sensitive data.** Whatever a local +variable holds — a password, a token, a whole request body — is written into +the log. Keep it off in production. If you do turn it on there, register +`RedactPlugin` and/or `PIIRedactPlugin`: every captured value is passed +through them *before* the traceback is formatted, so a local named `password` +or holding an email address is masked instead of printed. They only mask what +they're configured to recognize (a local's name, or a PII pattern in its +value) — a secret hiding inside a dict called `payload` is not caught. A +plugin whose redaction hook fails masks the value rather than showing it. + +```python +from logquill import Logger, RedactPlugin + +logger = Logger("app", plugins=[RedactPlugin()]) + + +def login(user, password): + raise PermissionError("bad credentials") + + +try: + login("ada", "hunter2") +except PermissionError: + record = logger.error("login failed", diagnose=True) + +assert "user = 'ada'" in record["meta"]["stack"] +assert "hunter2" not in record["meta"]["stack"] +assert "password = ***" in record["meta"]["stack"] +``` + `LogQuillHandler` bridges stdlib `logging` calls — including from third-party libraries you don't control — into a `Logger`, so they flow through the same transports and plugins instead of needing every call site @@ -882,6 +1029,110 @@ for _ in range(100): logger.error("connection refused") # only the first 5 per minute ship ``` +## Lazy values, caller depth & disabling a library + +**Lazy values.** A `DEBUG`/`TRACE` call left in production code still builds +its arguments before the logger discards it. `logger.opt(lazy=True)` defers +any callable `meta` value until the record is really going to be emitted, so +a filtered call costs nothing. (If a callable raises, the record carries a +placeholder naming the error; the exception never reaches your code.) + +```python +from logquill import Logger + +logger = Logger("app", level="INFO") +calls = [] + + +def expensive_dump(): + calls.append(1) + return {"rows": 100_000} + + +logger.opt(lazy=True).debug("state", dump=expensive_dump) # filtered: never called +assert calls == [] + +record = logger.opt(lazy=True).info("state", dump=expensive_dump) # emitted: called once +assert calls == [1] +assert record["meta"]["dump"] == {"rows": 100_000} +``` + +**Caller depth.** `logger.opt(depth=N)` adds `meta.caller` (`module`, +`function`, `line`, `file`) naming the code that logged, `N` frames up from the +direct caller. Use it in a wrapper or decorator so the record points at the +wrapper's caller instead of the wrapper itself: + +```python +from logquill import Logger + +logger = Logger("app") + + +def audit(message): + return logger.opt(depth=1).info(message) # report audit()'s caller, not audit() + + +def transfer_funds(): + return audit("funds transferred") + + +record = transfer_funds() +assert record["meta"]["caller"]["function"] == "transfer_funds" +``` + +**Disabling a library.** When a library uses LogQuill internally, it should +be silent in its host application by default. The library calls +`logquill.disable(__name__)` once at import; the application can opt back in +with `enable()`. Rules cover a logger and everything nested under it, and the +most specific rule wins: + +```python +import logquill +from logquill import Logger + +library_log = Logger("mylib.http") # what a library `mylib` would create + +logquill.disable("mylib") +assert library_log.info("hidden") is None + +logquill.enable("mylib.http") # the app wants just this part back +assert library_log.info("visible") is not None +``` + +## Parsing log files + +`parse()` extracts structured fields from a log file with a regex — including +logs LogQuill didn't write, like a legacy app's or a third-party tool's. It +yields a dict of the pattern's named groups for every matching line and skips +the rest. It reads one line at a time, so a multi-gigabyte file costs no more +memory than a small one. `cast` converts groups as they're read: + +```python +import tempfile +from pathlib import Path + +from logquill import parse + +with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "legacy.log" + path.write_text( + "2026-01-01 10:00:00 [INFO] 200 started\n" + "not a log line\n" + "2026-01-01 10:00:05 [ERROR] 503 upstream down\n" + ) + + pattern = r"(?P\S+ \S+) \[(?P[A-Z]+)\] (?P\d+) (?P.*)" + errors = [e for e in parse(path, pattern, cast={"code": int}) if e["level"] == "ERROR"] + +assert errors == [ + {"when": "2026-01-01 10:00:05", "level": "ERROR", "code": 503, "message": "upstream down"} +] +``` + +To read back what `TextFormatter` wrote, use the ready-made +`TEXT_LOG_PATTERN` with `cast=TEXT_LOG_CASTS` (which decodes `meta` from JSON); +for `LogfmtFormatter` output, `parse_logfmt(line)` returns a dict of strings. + ## CLI Installing `logquill` also installs a `logquill` command for local diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/benchmarks/measure.py b/benchmarks/measure.py new file mode 100644 index 0000000..7c02ccc --- /dev/null +++ b/benchmarks/measure.py @@ -0,0 +1,197 @@ +"""Memory measurements for the logging hot path, and the budgets CI holds them to. + +These measure bytes and object counts, not wall-clock time: allocation +numbers are close to deterministic for a given interpreter, so a budget can +fail the build on a real regression without flaking on a busy CI runner the +way a timing threshold would. + +Run them with `pytest benchmarks`. They're kept out of the default `pytest` +run (and out of the coverage job) because line tracing distorts allocation +counts. +""" + +from __future__ import annotations + +import gc +import statistics +import sys +import threading +import tracemalloc +from typing import Callable + +from logquill import ContextPlugin, Logger, PIIRedactPlugin, Plugin, RedactPlugin +from logquill.records import LogRecord +from logquill.transports.transport import Transport + +#: The most a metric may be before the build fails. Each is a few times what +#: the current code measures, so ordinary interpreter-to-interpreter variation +#: passes and a change that makes the hot path materially heavier doesn't. +BUDGETS: dict[str, float] = { + # net objects still alive after a log call, per call — a fixed few hundred + # one-off allocations amortize to ~0.01 over the run; anything near 1 means + # the hot path is accumulating state + "retained_blocks_per_call": 0.05, + # transient bytes at the peak of one call: level-filtered (never builds a record) + "peak_bytes_filtered_call": 1_000, + # ... a call through no plugins, and through a realistic plugin stack + "peak_bytes_plain_call": 12_000, + "peak_bytes_plugin_call": 14_000, + # memory growth over a 100k-record burst into a stalled transport, with a + # bounded queue — the queue bound, not the burst size, must set this + "peak_bytes_stalled_burst": 3_000_000, +} + +_META = {"user_id": 42, "route": "/checkout", "duration_ms": 12.5, "ok": True, "tag": "a" * 40} + + +class NullTransport(Transport): + """Discards everything: isolates the logger's own cost from any sink's.""" + + def write(self, formatted: str, record: LogRecord) -> None: + pass + + +class StalledTransport(Transport): + """Blocks in `write()` until released — a sink that's down.""" + + def __init__(self) -> None: + super().__init__() + self.gate = threading.Event() + self.started = threading.Event() + + def write(self, formatted: str, record: LogRecord) -> None: + self.started.set() + self.gate.wait(timeout=60) + + +def plain_logger(level: str = "INFO") -> Logger: + return Logger("bench", level=level, transports=[NullTransport()], flush_at_exit=False) + + +def plugin_logger() -> Logger: + return Logger( + "bench", + transports=[NullTransport()], + plugins=[ContextPlugin(service="api"), RedactPlugin(), PIIRedactPlugin()], + flush_at_exit=False, + ) + + +def retained_blocks_per_call(logger: Logger, calls: int = 20_000) -> float: + """Net change in live allocated objects per call, with the cyclic GC held + off so it can't mask a leak or add noise.""" + for _ in range(200): + logger.info("warmup", **_META) + gc.collect() + gc.disable() + try: + before = sys.getallocatedblocks() + for _ in range(calls): + logger.info("bench", **_META) + after = sys.getallocatedblocks() + finally: + gc.enable() + return (after - before) / calls + + +def peak_bytes_per_call(call: Callable[[], object], samples: int = 100) -> float: + """Median, over `samples` calls, of how far traced memory climbed above + its starting level during one call — the transient cost of a log call.""" + for _ in range(50): + call() + gc.collect() + peaks: list[int] = [] + for _ in range(samples): + tracemalloc.start() + try: + baseline, _peak = tracemalloc.get_traced_memory() + call() + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + peaks.append(peak - baseline) + return statistics.median(peaks) + + +def peak_bytes_stalled_burst( + *, burst: int = 100_000, max_queue_size: int = 1_000, policy: str = "drop_oldest" +) -> float: + """Memory growth while `burst` records go into a queue whose consumer is + stalled for the whole burst.""" + transport = StalledTransport() + logger = Logger( + "bench", + transports=[transport], + async_dispatch=True, + max_queue_size=max_queue_size, + backpressure=policy, # type: ignore[arg-type] + flush_at_exit=False, + ) + logger.info("stall the worker", **_META) + transport.started.wait(timeout=10) + gc.collect() + tracemalloc.start() + try: + baseline, _peak = tracemalloc.get_traced_memory() + for _ in range(burst): + logger.info("burst", **_META) + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + transport.gate.set() + logger.close(timeout=30) + return float(peak - baseline) + + +def measure_all() -> dict[str, float]: + """Every metric in `BUDGETS`, for the logger configurations the library ships.""" + filtered = plain_logger(level="ERROR") + plain = plain_logger() + with_plugins = plugin_logger() + return { + "retained_blocks_per_call": retained_blocks_per_call(plain), + "peak_bytes_filtered_call": peak_bytes_per_call(lambda: filtered.info("x", **_META)), + "peak_bytes_plain_call": peak_bytes_per_call(lambda: plain.info("x", **_META)), + "peak_bytes_plugin_call": peak_bytes_per_call(lambda: with_plugins.info("x", **_META)), + "peak_bytes_stalled_burst": peak_bytes_stalled_burst(), + } + + +def violations(metrics: dict[str, float], budgets: dict[str, float] = BUDGETS) -> list[str]: + """One human-readable line per metric that's over its budget (or missing).""" + problems = [] + for name, limit in budgets.items(): + value = metrics.get(name) + if value is None: + problems.append(f"{name}: not measured") + elif value > limit: + problems.append(f"{name}: {value:,.1f} is over the budget of {limit:,.1f}") + return problems + + +class Leaky(Plugin): + """A deliberately regressed plugin, used to prove the gate can fail: it + keeps every record it sees, so memory grows with every call.""" + + def __init__(self) -> None: + self.seen: list[LogRecord] = [] + + def before_log(self, record: LogRecord) -> LogRecord | None: + self.seen.append(record) + return record + + +class Bloated(Plugin): + """Another deliberate regression: allocates a large transient buffer per call.""" + + def before_log(self, record: LogRecord) -> LogRecord | None: + record["meta"]["scratch"] = ["x" * 100 for _ in range(2_000)] + return record + + +if __name__ == "__main__": # pragma: no cover + results = measure_all() + for name, value in results.items(): + print(f"{name:32} {value:>14,.1f} (budget {BUDGETS[name]:g})") + problems = violations(results) + sys.exit("\n".join(problems) if problems else 0) diff --git a/benchmarks/test_budgets.py b/benchmarks/test_budgets.py new file mode 100644 index 0000000..fbe26ae --- /dev/null +++ b/benchmarks/test_budgets.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import pytest + +from benchmarks import measure + + +def test_hot_path_memory_stays_within_budget() -> None: + results = measure.measure_all() + + problems = measure.violations(results) + + assert not problems, "memory budget exceeded:\n " + "\n ".join(problems) + + +def test_the_gate_fails_on_a_seeded_regression() -> None: + """A budget that can't fail protects nothing: feed the gate deliberately + regressed code and confirm it reports it.""" + leaky = measure.Leaky() + bloated = measure.Bloated() + leaking_logger = measure.plain_logger() + leaking_logger.use(leaky) + bloated_logger = measure.plain_logger() + bloated_logger.use(bloated) + + results = { + "retained_blocks_per_call": measure.retained_blocks_per_call(leaking_logger, calls=2_000), + "peak_bytes_plain_call": measure.peak_bytes_per_call( + lambda: bloated_logger.info("x", **measure._META) + ), + # an "unbounded" queue is exactly the regression the bound exists to prevent + "peak_bytes_stalled_burst": measure.peak_bytes_stalled_burst( + burst=20_000, max_queue_size=1_000_000 + ), + } + + problems = measure.violations(results, {k: measure.BUDGETS[k] for k in results}) + assert len(problems) == 3, problems + + +def test_a_missing_metric_is_a_failure_not_a_silent_pass() -> None: + assert measure.violations({}) == [f"{name}: not measured" for name in measure.BUDGETS] + + +@pytest.mark.parametrize("policy", ["drop_oldest", "drop_newest"]) +def test_stalled_burst_memory_is_set_by_the_queue_bound_not_the_burst_size(policy: str) -> None: + small = measure.peak_bytes_stalled_burst(burst=20_000, max_queue_size=500, policy=policy) + large = measure.peak_bytes_stalled_burst(burst=100_000, max_queue_size=500, policy=policy) + + # 5x the burst must not cost anywhere near 5x the memory + assert large < small * 2 diff --git a/logquill/__init__.py b/logquill/__init__.py index e356c33..6a32c77 100644 --- a/logquill/__init__.py +++ b/logquill/__init__.py @@ -2,11 +2,14 @@ from logquill.config import load_config, logger_from_env, logger_from_file from logquill.context import bind_context, current_context from logquill.exceptions import format_exc_info -from logquill.formatter import Formatter, JSONFormatter +from logquill.formatters import Formatter, JSONFormatter, LogfmtFormatter, TextFormatter from logquill.handler import LogQuillHandler from logquill.levels import Level, parse_level from logquill.logger import Logger +from logquill.opt import OptLogger +from logquill.parsing import TEXT_LOG_CASTS, TEXT_LOG_PATTERN, parse, parse_logfmt from logquill.plugins.alerting_plugin import AlertingPlugin +from logquill.plugins.apprise_alert_plugin import AppriseAlertPlugin from logquill.plugins.context_plugin import ContextPlugin from logquill.plugins.email_alert_plugin import EmailAlertPlugin from logquill.plugins.pagerduty_alert_plugin import PagerDutyAlertPlugin @@ -21,6 +24,7 @@ from logquill.plugins.trace_context_plugin import TraceContextPlugin from logquill.records import LogRecord from logquill.serverless import with_azure_function, with_cloud_function, with_lambda +from logquill.toggle import disable, enable, is_enabled from logquill.transports.batching_transport import BatchingTransport from logquill.transports.cloud.app_insights_transport import AppInsightsTransport from logquill.transports.cloud.cloud_logging_transport import CloudLoggingTransport @@ -51,6 +55,7 @@ __all__ = [ "AlertingPlugin", + "AppriseAlertPlugin", "AppInsightsTransport", "AsyncWorker", "BaseQueueTransport", @@ -72,6 +77,7 @@ "JSONFormatter", "KafkaTransport", "Level", + "LogfmtFormatter", "LogQuillAdapter", "LogQuillHandler", "LogRecord", @@ -79,6 +85,7 @@ "MongoDBTransport", "MySQLTransport", "NewRelicTransport", + "OptLogger", "PIIRedactPlugin", "PagerDutyAlertPlugin", "Plugin", @@ -95,16 +102,24 @@ "SamplingPlugin", "SlackAlertPlugin", "SyslogTransport", + "TEXT_LOG_CASTS", + "TEXT_LOG_PATTERN", "TamperEvidentPlugin", + "TextFormatter", "TraceContextPlugin", "Transport", "bind_context", "current_context", + "disable", + "enable", "format_exc_info", + "is_enabled", "load_config", "logger_from_env", "logger_from_file", + "parse", "parse_level", + "parse_logfmt", "with_azure_function", "with_cloud_function", "with_lambda", diff --git a/logquill/cli.py b/logquill/cli.py index a3b04a2..d0ee650 100644 --- a/logquill/cli.py +++ b/logquill/cli.py @@ -10,6 +10,7 @@ from pathlib import Path from typing import IO, Any, Sequence +from logquill.formatters import format_text from logquill.levels import Level, parse_level _COLORS = { @@ -93,20 +94,11 @@ def _parse_line(line: str, *, warn_stream: IO[str]) -> dict[str, Any] | None: def _format_human(record: dict[str, Any], *, colorize: bool) -> str: - level_name = str(record.get("level", "?")) - timestamp = record.get("timestamp", "?") - logger_name = record.get("logger", "?") - message = record.get("message", "") - meta = record.get("meta") or {} - - line = f"{timestamp} {level_name:<5} {logger_name}: {message}" - if meta: - line += f" {json.dumps(meta, separators=(',', ':'), default=str)}" - + line = format_text(record) if colorize: try: - color = _COLORS.get(parse_level(level_name)) - except (TypeError, ValueError): + color = _COLORS.get(parse_level(str(record.get("level", "?")))) + except ValueError: color = None if color: line = f"{color}{line}{_RESET}" diff --git a/logquill/config.py b/logquill/config.py index 945d80b..71c748e 100644 --- a/logquill/config.py +++ b/logquill/config.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Any +from logquill.formatters import Formatter, JSONFormatter, LogfmtFormatter, TextFormatter from logquill.logger import Logger from logquill.plugins.context_plugin import ContextPlugin from logquill.plugins.pii_redact_plugin import PIIRedactPlugin @@ -44,6 +45,15 @@ } +#: Built-in formatters selectable by name in a transport's `options` +#: (`{"formatter": "logfmt"}`), since a config file can't hold an instance. +_FORMATTER_TYPES: dict[str, type[Formatter]] = { + "json": JSONFormatter, + "text": TextFormatter, + "logfmt": LogfmtFormatter, +} + + def _resolve_class(entry: dict[str, Any], registry: dict[str, type]) -> type: if "class" in entry: dotted = entry["class"] @@ -76,10 +86,23 @@ def _build(entries: list[dict[str, Any]] | None, registry: dict[str, type]) -> l built = [] for entry in entries or []: cls = _resolve_class(entry, registry) - built.append(cls(**entry.get("options", {}))) + built.append(cls(**_resolve_options(entry.get("options", {})))) return built +def _resolve_options(options: dict[str, Any]) -> dict[str, Any]: + formatter = options.get("formatter") + if not isinstance(formatter, str): + return options + try: + return {**options, "formatter": _FORMATTER_TYPES[formatter]()} + except KeyError: + known = ", ".join(sorted(_FORMATTER_TYPES)) + raise ValueError( + f"Unknown formatter {formatter!r} — built-in formatters are: {known}." + ) from None + + def load_config(data: dict[str, Any], *, name: str = "app") -> Logger: """Build a `Logger` from an already-parsed config dict — the same shape `logger_from_file`/`logger_from_env` parse JSON/YAML into: @@ -99,9 +122,14 @@ def load_config(data: dict[str, Any], *, name: str = "app") -> Logger: ], "async_dispatch": true, "max_queue_size": 10000, - "backpressure": "drop_oldest" + "backpressure": "drop_oldest", + "flush_at_exit": true } + A transport's `options` may name a built-in formatter as a string — + `{"type": "console", "options": {"formatter": "text"}}` — one of + `"json"` (the default), `"text"` or `"logfmt"`. + Each transport/plugin entry needs either `"type"` (a built-in shortcut — see the module-level registries above) or `"class"` (a fully-qualified dotted path, imported and instantiated the way `logging.config. @@ -109,7 +137,7 @@ def load_config(data: dict[str, Any], *, name: str = "app") -> Logger: this with config you trust, the same as any other deployment config). `"options"` becomes that class's constructor keyword arguments. - `"async_dispatch"`/`"max_queue_size"`/`"backpressure"` are optional and + `"async_dispatch"`/`"max_queue_size"`/`"backpressure"`/`"flush_at_exit"` are optional and map directly onto `Logger`'s constructor arguments of the same name — see there for what each does. """ @@ -125,6 +153,7 @@ def load_config(data: dict[str, Any], *, name: str = "app") -> Logger: async_dispatch=data.get("async_dispatch", False), max_queue_size=data.get("max_queue_size", 10_000), backpressure=data.get("backpressure", "drop_oldest"), + flush_at_exit=data.get("flush_at_exit", True), ) diff --git a/logquill/exceptions.py b/logquill/exceptions.py index 29d7c5e..df699cc 100644 --- a/logquill/exceptions.py +++ b/logquill/exceptions.py @@ -1,9 +1,26 @@ from __future__ import annotations +import logging import sys import traceback from types import TracebackType -from typing import Literal, Union +from typing import Callable, Literal, Union + +_logger = logging.getLogger("logquill") + +#: `(local variable name, its repr) -> the text to show instead` — how +#: `diagnose` mode gives redaction plugins a chance to mask a captured local +#: before it's ever formatted into the traceback. +LocalRedactor = Callable[[str, str], str] + +#: Longest repr shown for one captured local; the rest is cut with `...`. +_MAX_LOCAL_REPR = 200 + +#: A module-level frame's "locals" are the module's globals — mostly imported +#: modules and function/class definitions, which are noise, not state. +_NOISE_PREFIXES = (" str | None: +def format_exc_info( + exc_info: ExcInfoArg, + *, + diagnose: bool = False, + redact_local: LocalRedactor | None = None, +) -> str | None: """Render `exc_info` as a formatted traceback string, or `None` if there's nothing to format. Accepts the same shapes stdlib `logging` does, so `logger.error("failed", exc_info=e)` reads exactly like the @@ -27,6 +49,15 @@ def format_exc_info(exc_info: ExcInfoArg) -> str | None: - an exception instance — format it and its own traceback - an explicit `(type, value, traceback)` tuple - falsy (`False`/`None`, the default) — nothing to format + + `diagnose=True` also prints each frame's local variables under its + source line, which makes a failure far easier to debug — and can leak + secrets, since whatever a local holds (a password, a token, a whole + request body) lands in the log. Leave it off in production. When it is + on, every captured local is passed through `redact_local` *before* the + traceback is formatted, so a redaction plugin's rules apply to it; + `Logger` wires that up from its plugins. Without a `redact_local`, values + are shown as-is. """ if not exc_info: return None @@ -44,4 +75,64 @@ def format_exc_info(exc_info: ExcInfoArg) -> str | None: else: exc_type, exc_value, exc_tb = exc_info + if diagnose and exc_value is not None: + _warn_diagnose_once() + try: + return _format_with_locals(exc_type, exc_value, exc_tb, redact_local) + except Exception: + # a local whose `__repr__` raises, say — fall back to the plain + # traceback, which shows no values and so can't leak any + _logger.debug("diagnose: couldn't capture locals, using a plain traceback") + return "".join(traceback.format_exception(exc_type, exc_value, exc_tb)) + + +def _warn_diagnose_once() -> None: + global _diagnose_warned + if not _diagnose_warned: + _diagnose_warned = True + _logger.warning( + "diagnose=True writes local variable values into log records, which can leak " + "sensitive data (passwords, tokens, personal data) — keep it off in production, " + "and use RedactPlugin/PIIRedactPlugin if you must run it there" + ) + + +def _format_with_locals( + exc_type: type[BaseException], + exc_value: BaseException, + exc_tb: TracebackType | None, + redact_local: LocalRedactor | None, +) -> str: + formatted = traceback.TracebackException(exc_type, exc_value, exc_tb, capture_locals=True) + _scrub_locals(formatted, redact_local, set()) + return "".join(formatted.format()) + + +def _scrub_locals( + formatted: traceback.TracebackException, + redact_local: LocalRedactor | None, + seen: set[int], +) -> None: + """Redact and trim the locals captured on `formatted` and on every + exception chained to it, in place — before anything is formatted.""" + if id(formatted) in seen: + return + seen.add(id(formatted)) + + for frame in formatted.stack: + captured = frame.locals or {} + kept: dict[str, str] = {} + for name, text in captured.items(): + if name.startswith("__") or text.startswith(_NOISE_PREFIXES): + continue + if redact_local is not None: + text = redact_local(name, text) + kept[name] = text if len(text) <= _MAX_LOCAL_REPR else text[:_MAX_LOCAL_REPR] + "..." + frame.locals = kept + + chained = [formatted.__cause__, formatted.__context__] + chained.extend(getattr(formatted, "exceptions", None) or []) # exception groups (3.11+) + for other in chained: + if other is not None: + _scrub_locals(other, redact_local, seen) diff --git a/logquill/formatter.py b/logquill/formatter.py index 0731be6..a91099d 100644 --- a/logquill/formatter.py +++ b/logquill/formatter.py @@ -1,23 +1,9 @@ -from __future__ import annotations +"""Compatibility alias: the formatters now live in `logquill.formatters`. -import json -from typing import Protocol +Kept so `from logquill.formatter import Formatter, JSONFormatter` — the +import path used since 1.0.0 — keeps working. +""" -from logquill.records import LogRecord +from logquill.formatters import Formatter, JSONFormatter - -class Formatter(Protocol): - """`format(record) -> string`, per the transport contract shared with logquill-js.""" - - def format(self, record: LogRecord) -> str: - """Render `record` to the string a transport will write.""" - ... - - -class JSONFormatter: - """Serializes a record to the canonical JSON line shape.""" - - def format(self, record: LogRecord) -> str: - """Serializes `record` to a single compact JSON line; non-JSON-native - values fall back to `str()` rather than raising.""" - return json.dumps(record, separators=(",", ":"), default=str) +__all__ = ["Formatter", "JSONFormatter"] diff --git a/logquill/formatters/__init__.py b/logquill/formatters/__init__.py new file mode 100644 index 0000000..e37bd21 --- /dev/null +++ b/logquill/formatters/__init__.py @@ -0,0 +1,6 @@ +from logquill.formatters.base import Formatter +from logquill.formatters.json_formatter import JSONFormatter +from logquill.formatters.logfmt_formatter import LogfmtFormatter +from logquill.formatters.text_formatter import TextFormatter, format_text + +__all__ = ["Formatter", "JSONFormatter", "LogfmtFormatter", "TextFormatter", "format_text"] diff --git a/logquill/formatters/base.py b/logquill/formatters/base.py new file mode 100644 index 0000000..3349d01 --- /dev/null +++ b/logquill/formatters/base.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +from typing import Protocol + +from logquill.records import LogRecord + + +class Formatter(Protocol): + """`format(record) -> string`, per the transport contract shared with logquill-js.""" + + def format(self, record: LogRecord) -> str: + """Render `record` to the string a transport will write.""" + ... diff --git a/logquill/formatters/json_formatter.py b/logquill/formatters/json_formatter.py new file mode 100644 index 0000000..3bf2d13 --- /dev/null +++ b/logquill/formatters/json_formatter.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +import json + +from logquill.records import LogRecord + + +class JSONFormatter: + """Serializes a record to the canonical JSON line shape.""" + + def format(self, record: LogRecord) -> str: + """Serializes `record` to a single compact JSON line; non-JSON-native + values fall back to `str()` rather than raising.""" + return json.dumps(record, separators=(",", ":"), default=str) diff --git a/logquill/formatters/logfmt_formatter.py b/logquill/formatters/logfmt_formatter.py new file mode 100644 index 0000000..9ae6ed4 --- /dev/null +++ b/logquill/formatters/logfmt_formatter.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from typing import Any + +from logquill.records import LogRecord + +#: The core record fields, always emitted first. A `meta` key with one of +#: these names is emitted as `meta.` so it can't shadow the field the +#: record itself carries. +_RESERVED_KEYS = ("timestamp", "level", "logger", "message") + +#: Nested `meta` dicts are flattened into dotted keys (`http.status=200`) up +#: to this depth; anything deeper is emitted as one JSON string, so a +#: pathologically nested (or circular) value can't blow up the line. +_MAX_FLATTEN_DEPTH = 5 + +_CONTROL = "\\x00-\\x1f\\x7f\\x85\\u2028\\u2029" +_NEEDS_QUOTES = re.compile(rf'[\s"=\\{_CONTROL}]|^$') +_BAD_KEY_CHARS = re.compile(rf'[\s"=\\{_CONTROL}]') +_CONTROL_CHAR = re.compile(f"[{_CONTROL}]") +_ESCAPES = {'"': '\\"', "\\": "\\\\", "\n": "\\n", "\r": "\\r", "\t": "\\t"} + + +def _quote(text: str) -> str: + if not _NEEDS_QUOTES.search(text): + return text + escaped = "".join( + _ESCAPES.get(char) or (f"\\u{ord(char):04x}" if _CONTROL_CHAR.match(char) else char) + for char in text + ) + return f'"{escaped}"' + + +def _key(raw: object) -> str: + return _BAD_KEY_CHARS.sub("_", str(raw)) or "_" + + +def _scalar(value: Any) -> str: + if value is None: + return "null" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (int, float)): + return str(value) + if isinstance(value, str): + return _quote(value) + if isinstance(value, (list, tuple, dict)): + return _quote(_json(value)) + return _quote(_safe_str(value)) + + +def _json(value: Any) -> str: + try: + return json.dumps(value, separators=(",", ":"), default=str) + except Exception: + return _safe_str(value) + + +def _safe_str(value: Any) -> str: + try: + return str(value) + except Exception: + return f"" + + +def _flatten(prefix: str, value: Any, depth: int, out: list[str]) -> None: + if isinstance(value, Mapping) and value and depth < _MAX_FLATTEN_DEPTH: + for child_key, child in value.items(): + _flatten(f"{prefix}.{_key(child_key)}", child, depth + 1, out) + return + out.append(f"{prefix}={_scalar(value)}") + + +class LogfmtFormatter: + """Single-line `key=value` output — the logfmt convention popularized by + Heroku and Go's logging ecosystem — for teams whose downstream tooling + (Loki, Splunk, `grep`) expects it: + + timestamp=2026-01-01T00:00:00.000Z level=INFO logger=app message="user signed up" user_id=42 + + `meta` keys are emitted as top-level pairs after the four record fields; + nested dicts flatten to dotted keys (`http.status=200`); lists are + emitted as a JSON string. Values containing whitespace, `=`, quotes, or + control characters are double-quoted and escaped, so the output is + always exactly one line — a multi-line traceback in `meta.stack` becomes + a single quoted value with `\\n` escapes. `logquill.parse_logfmt` reads + the format back. + + Never raises on odd `meta` values; non-serializable objects degrade to + their `str()`. + """ + + def format(self, record: LogRecord) -> str: + """Renders `record` as one logfmt line.""" + pairs = [ + f"timestamp={_quote(str(record['timestamp']))}", + f"level={_quote(str(record['level']))}", + f"logger={_quote(str(record['logger']))}", + f"message={_quote(str(record['message']))}", + ] + for meta_key, value in record["meta"].items(): + key = _key(meta_key) + if key in _RESERVED_KEYS: + key = f"meta.{key}" + _flatten(key, value, 1, pairs) + return " ".join(pairs) diff --git a/logquill/formatters/text_formatter.py b/logquill/formatters/text_formatter.py new file mode 100644 index 0000000..0746b31 --- /dev/null +++ b/logquill/formatters/text_formatter.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import json +from collections.abc import Mapping +from typing import Any + +from logquill.records import LogRecord + + +def format_text(record: Mapping[str, Any]) -> str: + """Render a record as one human-readable entry: + + 2026-01-01T00:00:00.000Z INFO app.api: user signed up {"user_id":42} + + Tolerant of partial records (a missing field renders as `?`), which is + what lets the `logquill tail` CLI share this with `TextFormatter` for + log lines written by other tools. A formatted traceback in + `meta["stack"]` is printed on the lines after the entry, as a + traceback normally reads, instead of as one long escaped JSON string. + """ + meta = dict(record.get("meta") or {}) + stack = meta.get("stack") + if isinstance(stack, str) and stack: + del meta["stack"] + else: + stack = None + + line = ( + f"{record.get('timestamp', '?')} {str(record.get('level', '?')):<5} " + f"{record.get('logger', '?')}: {record.get('message', '')}" + ) + if meta: + line += f" {_dump_meta(meta)}" + if stack is not None: + line += "\n" + stack.rstrip("\n") + return line + + +def _dump_meta(meta: Mapping[str, Any]) -> str: + try: + return json.dumps(meta, separators=(",", ":"), default=str) + except Exception: + # e.g. a circular reference or a value whose `__str__` raises: keep + # the entry readable rather than let a log call's formatting fail + return json.dumps({key: _safe_repr(value) for key, value in meta.items()}, default=str) + + +def _safe_repr(value: Any) -> str: + try: + return repr(value)[:200] + except Exception: + return f"" + + +class TextFormatter: + """Human-readable single-entry output for local development and + terminals — `ConsoleTransport(formatter=TextFormatter())`. + + Use `JSONFormatter` (the default) for anything a machine will read; this + one is for eyes. It never raises on odd `meta` values (non-serializable + objects, circular references) — they degrade to their `repr()`. + """ + + def format(self, record: LogRecord) -> str: + """Renders `record` via `format_text`.""" + return format_text(record) diff --git a/logquill/logger.py b/logquill/logger.py index 02c933b..725ca77 100644 --- a/logquill/logger.py +++ b/logquill/logger.py @@ -4,13 +4,16 @@ import logging from typing import Any +from logquill import shutdown from logquill.context import current_context -from logquill.exceptions import format_exc_info +from logquill.exceptions import LocalRedactor, format_exc_info from logquill.levels import Level, parse_level +from logquill.opt import OptLogger, caller_info, resolve_lazy from logquill.plugins.context_plugin import ContextPlugin from logquill.plugins.plugin import FunctionPlugin, MiddlewareFunc, Plugin from logquill.records import LogRecord, create_record from logquill.span import SpanContext, current_span_id +from logquill.toggle import is_enabled from logquill.transports.transport import Transport from logquill.worker import AsyncWorker, BackpressurePolicy @@ -35,6 +38,7 @@ def __init__( async_dispatch: bool = False, max_queue_size: int = 10_000, backpressure: BackpressurePolicy = "drop_oldest", + flush_at_exit: bool = True, ) -> None: """`async_dispatch=True` moves per-record transport writes (and the `after_log` plugin hooks that follow them) onto a background thread, @@ -47,6 +51,15 @@ def __init__( `max_queue_size`/`backpressure` are only meaningful with `async_dispatch=True` — see `AsyncWorker` for what each `backpressure` policy does under a sustained burst. + + `flush_at_exit=True` (the default) drains queued records and closes + this logger's transports when the interpreter exits, so a script that + ends without calling `close()` doesn't lose its last records or a + batching transport's unsent batch. Pass `False` if you manage + shutdown yourself. It runs on a normal exit (end of script, + `sys.exit()`, an unhandled exception) but not when the process is + killed outright (`SIGKILL`, `os._exit()`, or `SIGTERM` with no + handler installed) — handle `SIGTERM` and call `close()` for that. """ self.name = name self._level = parse_level(level) @@ -59,6 +72,8 @@ def __init__( if async_dispatch else None ) + if flush_at_exit: + shutdown.register(self) @property def level(self) -> Level: @@ -141,6 +156,32 @@ def close(self, timeout: float | None = 5.0) -> None: self._worker.close(timeout) for transport in self.transports: transport.close() + shutdown.mark_closed(transport) + + def opt(self, *, lazy: bool = False, depth: int | None = None) -> OptLogger: + """A view of this logger with per-call options — `lazy=True` to + defer expensive `meta` values until the record is really emitted, + `depth=N` to report the caller `N` frames up the stack. See + `OptLogger`. + """ + return OptLogger(self, lazy=lazy, depth=depth) + + def _local_redactor(self) -> LocalRedactor: + """Chain every plugin's `redact_local` into one function for + `diagnose` mode. A plugin whose hook raises fails closed: the value + is masked rather than shown, since the alternative is leaking exactly + what that plugin exists to hide.""" + plugins = list(self.plugins) + + def redact(name: str, text: str) -> str: + for plugin in plugins: + try: + text = plugin.redact_local(name, text) + except Exception: + return "" + return text + + return redact def _notify_error(self, plugin: Plugin, exc: Exception, record: LogRecord) -> None: # a broken error handler must not crash logging either @@ -167,12 +208,45 @@ def _dispatch(self, record: LogRecord) -> None: except Exception as exc: self._notify_error(plugin, exc, record) - def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord | None: - if level < self._level: + def _log( + self, + level: Level, + message: str, + meta: dict[str, Any], + *, + lazy: bool = False, + depth: int | None = None, + ) -> LogRecord | None: + if level < self._level or not is_enabled(self.name): return None + if lazy: + meta = resolve_lazy(meta) + + if depth is not None: + caller = caller_info(depth) + if caller is not None: + meta.setdefault("caller", caller) + + diagnose = bool(meta.pop("diagnose", False)) + if diagnose: + meta.setdefault("exc_info", True) if "exc_info" in meta: - stack = format_exc_info(meta.pop("exc_info")) + exc_info = meta.pop("exc_info") + try: + stack = format_exc_info( + exc_info, + diagnose=diagnose, + redact_local=self._local_redactor() if diagnose else None, + ) + except Exception: + # a malformed value must not crash the caller that's just logging + _logger.warning( + "Logger: ignoring exc_info=%r — it must be True, an exception instance, " + "or a (type, value, traceback) tuple", + exc_info, + ) + stack = None if stack is not None: meta["stack"] = stack @@ -236,7 +310,16 @@ def error(self, message: str, /, **meta: Any) -> LogRecord | None: tuple — the same shapes stdlib `logging` accepts) formats a traceback into `meta["stack"]` and is otherwise not kept in `meta` as-is, since a raw exception object isn't serializable. Every - `Logger` method accepts it, not just this one.""" + `Logger` method accepts it, not just this one. + + `diagnose=True` additionally writes each frame's local variable + values into that traceback (and implies `exc_info=True` if none was + given). **Off by default, and it can leak sensitive data** — + whatever a local holds (a password, a token, a request body) ends up + in the log, so keep it off in production. Locals are passed through + the registered `RedactPlugin`/`PIIRedactPlugin` before the traceback + is formatted, but that only masks what those plugins are configured + to recognize (by variable name, or by PII pattern in the value).""" return self._log(Level.ERROR, message, meta) def fatal(self, message: str, /, **meta: Any) -> LogRecord | None: diff --git a/logquill/opt.py b/logquill/opt.py new file mode 100644 index 0000000..145d9d0 --- /dev/null +++ b/logquill/opt.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import sys +from typing import TYPE_CHECKING, Any + +from logquill.levels import Level +from logquill.records import LogRecord + +if TYPE_CHECKING: + from logquill.logger import Logger + + +def resolve_lazy(meta: dict[str, Any]) -> dict[str, Any]: + """Replace every callable value in `meta` with what it returns. A callable + that raises must not crash the caller that's just trying to log, so it + becomes a placeholder naming the error instead.""" + resolved: dict[str, Any] = {} + for key, value in meta.items(): + if callable(value): + try: + value = value() + except Exception as exc: + value = f"" + resolved[key] = value + return resolved + + +def caller_info(depth: int) -> dict[str, Any] | None: + """Where the code that logged is, `depth` frames further up the stack. + + Frame 0 is this function, 1 is `Logger._log`, 2 is the `OptLogger` method + that called it, and 3 is the code that called that — `depth=0` — so every + `OptLogger` method must call `Logger._log` directly for this count to hold. + Returns `None` if the stack isn't `depth` frames deep. + """ + try: + frame = sys._getframe(3 + depth) + except ValueError: + return None + return { + "module": frame.f_globals.get("__name__"), + "function": frame.f_code.co_name, + "line": frame.f_lineno, + "file": frame.f_code.co_filename, + } + + +class OptLogger: + """A view of a `Logger` with per-call options applied, from + `Logger.opt(...)`. It has the same logging methods as `Logger`, and writes + through the logger it came from — same level, plugins and transports. + + - `lazy=True`: callable `meta` values are called only if the record is + actually going to be emitted, so an expensive value costs nothing when + the logger's level filters the call out: + + logger.opt(lazy=True).debug("state", dump=lambda: expensive_dump()) + + The callables run on the calling thread, before the record is queued for + any async dispatch, so they see the state at the moment of the call. + - `depth=N`: adds `meta.caller` (`module`, `function`, `line`, `file`) + naming the code that logged, `N` frames up from the direct caller. Use + it inside a wrapper or decorator so the record points at the wrapper's + caller instead of the wrapper: + + def audit(message): + logger.opt(depth=1).info(message) # reports audit()'s caller + """ + + def __init__(self, logger: Logger, *, lazy: bool = False, depth: int | None = None) -> None: + """See the class docstring. `depth` must be `None` (don't record a + caller) or a non-negative integer.""" + if depth is not None and depth < 0: + raise ValueError(f"opt(depth=...) must be >= 0, got {depth}") + self._logger = logger + self._lazy = lazy + self._depth = depth + + def trace(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.trace` with this view's options.""" + return self._logger._log(Level.TRACE, message, meta, lazy=self._lazy, depth=self._depth) + + def debug(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.debug` with this view's options.""" + return self._logger._log(Level.DEBUG, message, meta, lazy=self._lazy, depth=self._depth) + + def info(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.info` with this view's options.""" + return self._logger._log(Level.INFO, message, meta, lazy=self._lazy, depth=self._depth) + + def warn(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.warn` with this view's options.""" + return self._logger._log(Level.WARN, message, meta, lazy=self._lazy, depth=self._depth) + + def error(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.error` with this view's options.""" + return self._logger._log(Level.ERROR, message, meta, lazy=self._lazy, depth=self._depth) + + def fatal(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.fatal` with this view's options.""" + return self._logger._log(Level.FATAL, message, meta, lazy=self._lazy, depth=self._depth) + + def thought(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.thought` with this view's options.""" + return self._logger._log( + Level.INFO, message, {"kind": "thought", **meta}, lazy=self._lazy, depth=self._depth + ) + + def action(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.action` with this view's options.""" + return self._logger._log( + Level.INFO, message, {"kind": "action", **meta}, lazy=self._lazy, depth=self._depth + ) + + def observation(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.observation` with this view's options.""" + return self._logger._log( + Level.INFO, message, {"kind": "observation", **meta}, lazy=self._lazy, depth=self._depth + ) + + def decision(self, message: str, /, **meta: Any) -> LogRecord | None: + """`Logger.decision` with this view's options.""" + return self._logger._log( + Level.INFO, message, {"kind": "decision", **meta}, lazy=self._lazy, depth=self._depth + ) diff --git a/logquill/parsing.py b/logquill/parsing.py new file mode 100644 index 0000000..aeef772 --- /dev/null +++ b/logquill/parsing.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import contextlib +import json +import os +import re +from collections.abc import Callable, Iterable, Iterator, Mapping +from typing import Any + +#: Matches one entry written by `TextFormatter` (single-line entries; a +#: traceback printed on the lines after an entry isn't part of the match). +#: Pass with `cast=TEXT_LOG_CASTS` to get `meta` back as a dict: +#: +#: parse("app.log", TEXT_LOG_PATTERN, cast=TEXT_LOG_CASTS) +TEXT_LOG_PATTERN = ( + r"^(?P\S+) (?P[A-Z]+)\s+(?P[^:\s]+): " + r"(?P.*?)(?: (?P\{.*\}))?$" +) + +#: `cast` mapping that decodes the `meta` group of `TEXT_LOG_PATTERN` from JSON. +TEXT_LOG_CASTS: dict[str, Callable[[str], Any]] = {"meta": json.loads} + + +def parse( + source: str | os.PathLike[str] | Iterable[str], + pattern: str | re.Pattern[str], + *, + cast: Mapping[str, Callable[[str], Any]] | None = None, + encoding: str = "utf-8", +) -> Iterator[dict[str, Any]]: + r"""Extract structured fields from a log file with a regex, yielding one + dict per matching line — including logs LogQuill didn't write (legacy + apps, third-party tools), which is the reason to reach for this over + `json.loads`: + + pattern = r"(?P\S+ \S+) \[(?P[A-Z]+)\] (?P\d+) (?P.*)" + for entry in parse("legacy.log", pattern, cast={"code": int}): + if entry["level"] == "ERROR" and entry["code"] >= 500: + print(entry["when"], entry["message"]) + + `source` is a path (`str` or `os.PathLike`), or any iterable of lines + such as an open file or `sys.stdin`. `pattern` must contain at least one + named group, `(?P...)`; each yielded dict maps group names to the + matched text (`None` for an optional group that didn't participate). + Lines that don't match are skipped. `cast` maps group names to a + function applied to that group's text (`{"status": int}`); a group that + is `None` is left as is. + + Streams line by line, so memory stays flat however large the file is. + A pattern is matched against one line at a time, so it can't span + multiple lines (e.g. a wrapped traceback). + + Raises `ValueError` immediately for a pattern with no named groups or a + `cast` key that isn't one, and — while iterating — if a `cast` function + rejects a value, naming the line number so the entry can be found. + """ + compiled = re.compile(pattern) if isinstance(pattern, str) else pattern + if not compiled.groupindex: + raise ValueError( + "parse(): the pattern has no named groups — wrap the fields you want in " + "(?P...), e.g. r'(?P[A-Z]+) (?P.*)'" + ) + casts = dict(cast or {}) + unknown = sorted(set(casts) - set(compiled.groupindex)) + if unknown: + raise ValueError( + f"parse(): cast refers to {unknown}, which are not named groups in the " + f"pattern (groups: {sorted(compiled.groupindex)})" + ) + + return _scan(source, compiled, casts, encoding) + + +def _scan( + source: str | os.PathLike[str] | Iterable[str], + compiled: re.Pattern[str], + casts: Mapping[str, Callable[[str], Any]], + encoding: str, +) -> Iterator[dict[str, Any]]: + with contextlib.ExitStack() as stack: + if isinstance(source, (str, os.PathLike)): + lines: Iterable[str] = stack.enter_context(open(source, encoding=encoding)) + else: + lines = source + for number, line in enumerate(lines, start=1): + match = compiled.search(line.rstrip("\r\n")) + if match is None: + continue + fields: dict[str, Any] = match.groupdict() + for name, convert in casts.items(): + if fields[name] is None: + continue + try: + fields[name] = convert(fields[name]) + except Exception as exc: + raise ValueError( + f"parse(): line {number}: cast for {name!r} failed on " + f"{fields[name]!r} ({exc}) — make the cast tolerant of this value " + "or tighten the pattern so the group only matches what it can convert" + ) from exc + yield fields + + +_LOGFMT_PAIR = re.compile(r'([^\s="]+)(?:=("(?:[^"\\]|\\.)*"|[^\s"]*))?') +_UNESCAPES = {'"': '"', "\\": "\\", "n": "\n", "r": "\r", "t": "\t"} +_ESCAPE = re.compile(r"\\(u[0-9a-fA-F]{4}|.)", re.DOTALL) + + +def _unescape(match: re.Match[str]) -> str: + code = match.group(1) + if len(code) == 5: + return chr(int(code[1:], 16)) + return _UNESCAPES.get(code, "\\" + code) + + +def parse_logfmt(line: str) -> dict[str, str]: + """Split one logfmt line (`LogfmtFormatter`'s output, or any + Heroku/Go-style `key=value` line) into a dict of strings. Quoted values + are unescaped; a bare key with no `=` maps to `""`; if a key repeats, the + last value wins. Values are always strings — convert them yourself, or + pass a line through `parse()` with a pattern if you need typed fields. + """ + fields: dict[str, str] = {} + for key, value in _LOGFMT_PAIR.findall(line): + if value.startswith('"') and value.endswith('"') and len(value) >= 2: + value = _ESCAPE.sub(_unescape, value[1:-1]) + fields[key] = value + return fields diff --git a/logquill/plugins/apprise_alert_plugin.py b/logquill/plugins/apprise_alert_plugin.py new file mode 100644 index 0000000..6186a55 --- /dev/null +++ b/logquill/plugins/apprise_alert_plugin.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import threading +from collections.abc import Sequence +from typing import Any + +from logquill.plugins.alerting_plugin import AlertingPlugin +from logquill.records import LogRecord + +# Apprise's own notification types, which each service maps onto its native +# severity (a color, an emoji, a priority). +_NOTIFY_TYPES = {"WARN": "warning", "ERROR": "failure", "FATAL": "failure"} + + +class AppriseAlertPlugin(AlertingPlugin): + """Sends deduplicated `AlertingPlugin` alerts through + [Apprise](https://github.com/caronc/apprise), which speaks to 100+ + notification services (Discord, Telegram, Teams, ntfy, Matrix, SMS + gateways, ...) from one URL each — so "can it alert to X?" is answered + by Apprise's service list rather than a bespoke plugin per service. + + logger.use(AppriseAlertPlugin(["discord://webhook_id/webhook_token", + "ntfy://my-topic"])) + + Requires the optional dependency: `pip install logquill[apprise]`. + `SlackAlertPlugin` and `PagerDutyAlertPlugin` remain the better choice + for those two services, where a purpose-built plugin can format richer + messages than Apprise's generic title-and-body interface. + + Like every `AlertingPlugin`, alerts go out on a background thread and a + failure (an unreachable service, a rejected credential) is routed to + `on_error`, never raised into the code that logged. + """ + + def __init__( + self, + urls: str | Sequence[str], + *, + title: str | None = None, + apprise_client: Any = None, + **kwargs: Any, + ) -> None: + """`urls` is one Apprise service URL or a list of them. `title` + overrides the default `"[LEVEL] logger"`. `apprise_client` is any + object with Apprise's `notify(body=, title=, notify_type=)` method, + for tests or a client you've configured yourself (then `urls` is + ignored). `kwargs` go to `AlertingPlugin.__init__` (`threshold`, + `dedupe_window_seconds`, ...). Raises `ValueError` if Apprise rejects + a URL, so a typo fails at startup instead of on the first alert.""" + super().__init__(**kwargs) + self.title = title + self._notify_lock = threading.Lock() + self._client: Any = apprise_client if apprise_client is not None else _build_client(urls) + + def send_alert(self, record: LogRecord, occurrences: int) -> None: + """Sends one notification for `record` to every configured service; + raises if Apprise reports that delivery failed (caught by + `AlertingPlugin`'s `_safe_send` wrapper, so this never crashes the + caller).""" + body = record["message"] if occurrences <= 1 else f"{record['message']} (x{occurrences})" + title = self.title or f"[{record['level']}] {record['logger']}" + with self._notify_lock: + delivered = self._client.notify( + body=body, + title=title, + notify_type=_NOTIFY_TYPES.get(record["level"], "info"), + ) + if not delivered: + raise RuntimeError( + "AppriseAlertPlugin: Apprise reported that at least one notification failed to " + "send — check the service URLs and that the services are reachable; run " + "`apprise -vvv -b test ` to see the underlying error" + ) + + +def _build_client(urls: str | Sequence[str]) -> Any: + try: + import apprise + except ImportError as exc: + raise ImportError( + "AppriseAlertPlugin requires the optional `apprise` dependency — " + "install with `pip install logquill[apprise]`." + ) from exc + client = apprise.Apprise() + for url in [urls] if isinstance(urls, str) else urls: + if not client.add(url): + raise ValueError( + f"AppriseAlertPlugin: Apprise doesn't recognize the service URL {url!r} — " + "see https://github.com/caronc/apprise/wiki for each service's URL format" + ) + return client diff --git a/logquill/plugins/pii_redact_plugin.py b/logquill/plugins/pii_redact_plugin.py index cb9b259..d97bd04 100644 --- a/logquill/plugins/pii_redact_plugin.py +++ b/logquill/plugins/pii_redact_plugin.py @@ -125,3 +125,9 @@ def _redact_with_presidio(self, text: str) -> str: ) anonymized = self._anonymizer.anonymize(text=text, analyzer_results=results) return str(anonymized.text) + + def redact_local(self, name: str, text: str) -> str: + """Scrubs PII-shaped substrings out of the `repr` of a local variable + captured by `diagnose=True`, the same way `before_log` scrubs `meta` + values.""" + return self._redact_text(text) diff --git a/logquill/plugins/plugin.py b/logquill/plugins/plugin.py index a2f52a0..1bb4187 100644 --- a/logquill/plugins/plugin.py +++ b/logquill/plugins/plugin.py @@ -25,6 +25,16 @@ def after_log(self, record: LogRecord) -> None: def on_error(self, exc: Exception, record: LogRecord) -> None: """Called when one of this plugin's own hooks raises.""" + def redact_local(self, name: str, text: str) -> str: + """Optional hook for redaction plugins: called with the name and + `repr` of each local variable that `diagnose=True` captures into a + traceback, before the traceback is formatted, and returns the text to + show instead. Defaults to returning `text` unchanged. Only override + this if your plugin masks sensitive values — see `RedactPlugin`. If + it raises, the value is replaced with a placeholder rather than shown. + """ + return text + class FunctionPlugin(Plugin): """Wraps a plain `before_log`-style function as a `Plugin`. diff --git a/logquill/plugins/redact_plugin.py b/logquill/plugins/redact_plugin.py index 238f97c..66cc9df 100644 --- a/logquill/plugins/redact_plugin.py +++ b/logquill/plugins/redact_plugin.py @@ -30,3 +30,10 @@ def before_log(self, record: LogRecord) -> LogRecord | None: for key, value in meta.items() } return record + + def redact_local(self, name: str, text: str) -> str: + """Masks a local variable captured by `diagnose=True` when its name + matches `keys` (case-insensitively) — the same rule `before_log` + applies to `meta` keys, so `password`, `token` and friends never + reach a traceback either.""" + return self.replacement if name.lower() in self.keys else text diff --git a/logquill/shutdown.py b/logquill/shutdown.py new file mode 100644 index 0000000..5b11421 --- /dev/null +++ b/logquill/shutdown.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import atexit +import logging +import threading +import weakref +from typing import TYPE_CHECKING + +from logquill.transports.transport import Transport +from logquill.worker import AsyncWorker + +if TYPE_CHECKING: + from logquill.logger import Logger + +_logger = logging.getLogger("logquill") + +#: How long the exit hook waits for queued records to drain before giving up +#: and letting the process exit anyway — a stalled sink must not be able to +#: hang shutdown forever. +EXIT_DRAIN_TIMEOUT_SECONDS = 5.0 + +# Weak, so registering never keeps a logger (or its transports) alive. +_loggers: weakref.WeakSet[Logger] = weakref.WeakSet() +_closed_transports: weakref.WeakSet[Transport] = weakref.WeakSet() +_lock = threading.Lock() +_hook_installed = False + + +def register(logger: Logger) -> None: + """Have the process-exit hook flush and close `logger`'s worker and + transports if the process ends without `logger.close()` having been + called. Installs the single `atexit` hook on first use.""" + global _hook_installed + with _lock: + _loggers.add(logger) + if not _hook_installed: + atexit.register(shutdown) + _hook_installed = True + + +def mark_closed(transport: Transport) -> None: + """Record that `transport` was closed explicitly, so the exit hook + doesn't close it a second time (a child logger shares its parent's + transports).""" + with _lock: + _closed_transports.add(transport) + + +def shutdown(timeout: float = EXIT_DRAIN_TIMEOUT_SECONDS) -> None: + """Drain every registered logger's async queue (up to `timeout` + seconds), then close each transport once. Runs automatically at + interpreter exit; safe to call earlier, and safe to call twice. + + Never raises: anything that goes wrong is reported on the `logquill` + stdlib logger, because a failing flush must not turn a clean exit into + a traceback. + """ + with _lock: + loggers = list(_loggers) + already_closed = set(_closed_transports) + + workers: dict[int, AsyncWorker] = {} + transports: dict[int, Transport] = {} + for logger in loggers: + if logger._worker is not None: + workers[id(logger._worker)] = logger._worker + for transport in logger.transports: + if transport not in already_closed: + transports[id(transport)] = transport + + for worker in workers.values(): + try: + if not worker.close(timeout): + _logger.warning( + "logquill: %.1fs wasn't enough to flush every queued record at exit — " + "a transport is slow or unreachable; call logger.close(timeout=...) " + "yourself with a longer timeout if these records matter", + timeout, + ) + except Exception: + _logger.exception("logquill: failed to drain the async queue at exit") + + for transport in transports.values(): + try: + transport.close() + except Exception: + _logger.exception("%s: failed to close at exit", type(transport).__name__) + mark_closed(transport) diff --git a/logquill/toggle.py b/logquill/toggle.py new file mode 100644 index 0000000..2d7374c --- /dev/null +++ b/logquill/toggle.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import threading + +# Copy-on-write: writers swap in a new dict under the lock, readers just read +# the current reference. The hot path (`is_enabled` on every log call) takes +# no lock, and with no rules set it's a single truthiness check. +_rules: dict[str, bool] = {} +_lock = threading.Lock() + + +def disable(name: str = "") -> None: + """Turn off every `Logger` whose name is `name` or nested under it + (`"mylib"` covers `"mylib"` and `"mylib.http"`, not `"mylib2"`), so its + log calls become no-ops. `disable("")` turns off everything. + + This is for libraries that use LogQuill internally: call it once at + import so the library is silent by default instead of polluting its + host application's logs, and let the application opt back in: + + # mylib/__init__.py + import logquill + logquill.disable(__name__) + + # the application, if it wants mylib's logs + logquill.enable("mylib") + + The most specific rule wins: `disable("mylib")` followed by + `enable("mylib.http")` silences mylib except `mylib.http`. + """ + _set(name, False) + + +def enable(name: str = "") -> None: + """Undo `disable()` for `name` and everything nested under it. See + `disable()` for how rules combine.""" + _set(name, True) + + +def is_enabled(name: str) -> bool: + """Whether a logger called `name` currently emits records — i.e. the + most specific `enable()`/`disable()` rule covering it, or `True` if + none does.""" + rules = _rules + if not rules: + return True + candidate = name + while True: + verdict = rules.get(candidate) + if verdict is not None: + return verdict + if not candidate: + return True + candidate = candidate.rpartition(".")[0] + + +def _set(name: str, enabled: bool) -> None: + global _rules + if not isinstance(name, str): + raise TypeError( + f"disable()/enable() take a logger name such as __name__, got {type(name).__name__}" + ) + with _lock: + _rules = {**_rules, name: enabled} + + +def _reset() -> None: + """Drop every rule. For tests.""" + global _rules + with _lock: + _rules = {} diff --git a/logquill/transports/aiohttp_sender.py b/logquill/transports/aiohttp_sender.py new file mode 100644 index 0000000..da51832 --- /dev/null +++ b/logquill/transports/aiohttp_sender.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import asyncio +import contextlib +import threading +from collections.abc import Sequence +from typing import Any + + +class AiohttpSender: + """An `HTTPTransport` sender that POSTs batches with `aiohttp` + (`pip install logquill[http]`) instead of `urllib`. + + The reason to choose it is connection reuse: one `aiohttp.ClientSession` + lives for the transport's lifetime, so every batch after the first rides + an already-open, keep-alive connection instead of paying a new TCP/TLS + handshake — a real saving against an HTTPS collector that receives a + batch every few seconds. The session runs on a private event loop in a + background thread, so it works the same whether or not your application + has an event loop of its own. + + Calling it still blocks until the POST completes (or fails), exactly as + the `urllib` sender does — the non-blocking guarantee for a log call + comes from `Logger(async_dispatch=True)`, not from the sender. + """ + + def __init__( + self, + *, + timeout: float = 10.0, + headers: dict[str, str] | None = None, + ) -> None: + """`timeout` bounds each request, in seconds. `headers` are sent on + every request, on top of `Content-Type: application/x-ndjson`. + Raises `ImportError` right away if `aiohttp` isn't installed, rather + than on the first flush.""" + try: + import aiohttp # noqa: F401 + except ImportError as exc: + raise ImportError( + "HTTPTransport(backend='aiohttp') requires the optional `aiohttp` " + "dependency — install with `pip install logquill[http]`." + ) from exc + self.timeout = timeout + self._headers = {"Content-Type": "application/x-ndjson", **(headers or {})} + self._lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._session: Any = None + + def __call__(self, url: str, batch: Sequence[str]) -> None: + """POSTs `batch` to `url` as newline-delimited JSON and waits for the + response. Raises if the request fails or the server answers 4xx/5xx.""" + loop = self._ensure_loop() + body = "\n".join(batch).encode("utf-8") + future = asyncio.run_coroutine_threadsafe(self._post(url, body), loop) + try: + future.result(timeout=self.timeout + 5.0) + except BaseException: + future.cancel() + raise + + def close(self) -> None: + """Close the session and stop the background loop. Idempotent; a + later call starts a fresh session.""" + with self._lock: + loop, thread = self._loop, self._thread + self._loop = self._thread = None + if loop is None or thread is None: + return + if self._session is not None: + closing = asyncio.run_coroutine_threadsafe(self._session.close(), loop) + # shutting down; there's nothing useful to do with a failed close + with contextlib.suppress(Exception): + closing.result(timeout=5.0) + self._session = None + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5.0) + if not thread.is_alive(): + loop.close() + + def _ensure_loop(self) -> asyncio.AbstractEventLoop: + with self._lock: + if self._loop is None: + loop = asyncio.new_event_loop() + thread = threading.Thread( + target=self._run_loop, args=(loop,), name="logquill-aiohttp", daemon=True + ) + thread.start() + self._loop, self._thread = loop, thread + return self._loop + + @staticmethod + def _run_loop(loop: asyncio.AbstractEventLoop) -> None: + asyncio.set_event_loop(loop) + loop.run_forever() + + async def _post(self, url: str, body: bytes) -> None: + import aiohttp + + if self._session is None: + self._session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=self.timeout), headers=self._headers + ) + async with self._session.post(url, data=body) as response: + response.raise_for_status() + await response.read() diff --git a/logquill/transports/batching_transport.py b/logquill/transports/batching_transport.py index 541586c..863dfc0 100644 --- a/logquill/transports/batching_transport.py +++ b/logquill/transports/batching_transport.py @@ -5,7 +5,7 @@ from abc import abstractmethod from typing import Generic, Sequence, TypeVar, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.transport import Transport diff --git a/logquill/transports/cloud/app_insights_transport.py b/logquill/transports/cloud/app_insights_transport.py index 9fd0b6a..38bb460 100644 --- a/logquill/transports/cloud/app_insights_transport.py +++ b/logquill/transports/cloud/app_insights_transport.py @@ -4,7 +4,7 @@ import urllib.request from typing import Callable, Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/cloud/cloud_logging_transport.py b/logquill/transports/cloud/cloud_logging_transport.py index ca841a1..a984df3 100644 --- a/logquill/transports/cloud/cloud_logging_transport.py +++ b/logquill/transports/cloud/cloud_logging_transport.py @@ -2,7 +2,7 @@ from typing import Any, Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/cloud/cloudwatch_transport.py b/logquill/transports/cloud/cloudwatch_transport.py index ecb311d..b29c806 100644 --- a/logquill/transports/cloud/cloudwatch_transport.py +++ b/logquill/transports/cloud/cloudwatch_transport.py @@ -2,7 +2,7 @@ from typing import Any, Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/cloud/datadog_transport.py b/logquill/transports/cloud/datadog_transport.py index 170d2ec..a3cf6f0 100644 --- a/logquill/transports/cloud/datadog_transport.py +++ b/logquill/transports/cloud/datadog_transport.py @@ -4,7 +4,7 @@ import urllib.request from typing import Callable, Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/cloud/elasticsearch_transport.py b/logquill/transports/cloud/elasticsearch_transport.py index 251362e..dffb7f2 100644 --- a/logquill/transports/cloud/elasticsearch_transport.py +++ b/logquill/transports/cloud/elasticsearch_transport.py @@ -5,7 +5,7 @@ import urllib.request from typing import Callable, Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/cloud/new_relic_transport.py b/logquill/transports/cloud/new_relic_transport.py index fa2af7e..c1bae62 100644 --- a/logquill/transports/cloud/new_relic_transport.py +++ b/logquill/transports/cloud/new_relic_transport.py @@ -9,7 +9,7 @@ from email.utils import parsedate_to_datetime from typing import Callable, Dict, Literal, Sequence, TypedDict -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/cloud/syslog_transport.py b/logquill/transports/cloud/syslog_transport.py index ee9f756..4745c6f 100644 --- a/logquill/transports/cloud/syslog_transport.py +++ b/logquill/transports/cloud/syslog_transport.py @@ -4,7 +4,7 @@ import socket from typing import Callable -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.levels import Level, parse_level from logquill.records import LogRecord from logquill.transports.transport import Transport diff --git a/logquill/transports/console_transport.py b/logquill/transports/console_transport.py index 3bc75be..a631160 100644 --- a/logquill/transports/console_transport.py +++ b/logquill/transports/console_transport.py @@ -3,7 +3,7 @@ import sys from typing import TextIO -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.levels import Level, parse_level from logquill.records import LogRecord from logquill.transports.transport import Transport diff --git a/logquill/transports/file_transport.py b/logquill/transports/file_transport.py index a326980..e4786eb 100644 --- a/logquill/transports/file_transport.py +++ b/logquill/transports/file_transport.py @@ -3,7 +3,7 @@ from pathlib import Path from typing import Any, BinaryIO, TextIO, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.transport import Transport diff --git a/logquill/transports/http_transport.py b/logquill/transports/http_transport.py index 015e427..1fb2261 100644 --- a/logquill/transports/http_transport.py +++ b/logquill/transports/http_transport.py @@ -1,33 +1,47 @@ from __future__ import annotations +import logging import urllib.request -from typing import Callable, Sequence +from typing import Callable, Literal, Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.transport import Transport Sender = Callable[[str, Sequence[str]], None] +_logger = logging.getLogger("logquill") -def _urllib_sender(url: str, batch: Sequence[str]) -> None: - body = "\n".join(batch).encode("utf-8") - request = urllib.request.Request( - url, - data=body, - headers={"Content-Type": "application/x-ndjson"}, - method="POST", - ) - with urllib.request.urlopen(request, timeout=10) as response: # noqa: S310 - response.read() + +def _urllib_sender(timeout: float) -> Sender: + def send(url: str, batch: Sequence[str]) -> None: + body = "\n".join(batch).encode("utf-8") + request = urllib.request.Request( + url, + data=body, + headers={"Content-Type": "application/x-ndjson"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 + response.read() + + return send class HTTPTransport(Transport): """Batches formatted records and POSTs them as newline-delimited JSON. - Uses `urllib` (stdlib) by default so the core package stays dependency-free. - Pass `sender` to swap in a fake for tests, or a different backend (e.g. an - aiohttp-based one, once a non-blocking async dispatch path exists). + Uses `urllib` (stdlib) by default so the core package stays + dependency-free. `backend="aiohttp"` (`pip install logquill[http]`) sends + over a reused keep-alive connection instead — see `AiohttpSender`. Pass + `sender` to swap in a fake for tests, or any other backend. + + The buffer is bounded by both `batch_size` records and `max_bytes` of + formatted text — a flush fires as soon as either is reached — so a few + huge records can't grow it without limit. A failed send is logged and + that batch is dropped, never raised into the code that logged; the + transport keeps accepting records, so a down endpoint costs completeness, + not the process. """ def __init__( @@ -36,31 +50,66 @@ def __init__( *, formatter: Formatter | None = None, batch_size: int = 50, + max_bytes: int = 1_000_000, + timeout: float = 10.0, + backend: Literal["urllib", "aiohttp"] = "urllib", sender: Sender | None = None, ) -> None: - """`sender` defaults to a stdlib `urllib`-based POST; override for a - fake in tests or an alternate HTTP backend.""" + """`sender` overrides `backend` when given. `timeout` (seconds) bounds + each request for the built-in backends; it's ignored when you pass + your own `sender`. Raises `ValueError` for an unknown `backend`, and + `ImportError` for `backend="aiohttp"` without `aiohttp` installed.""" super().__init__(formatter) self.url = url self.batch_size = batch_size - self._sender: Sender = sender or _urllib_sender + self.max_bytes = max_bytes + self._sender: Sender + if sender is not None: + self._sender = sender + elif backend == "urllib": + self._sender = _urllib_sender(timeout) + elif backend == "aiohttp": + from logquill.transports.aiohttp_sender import AiohttpSender + + self._sender = AiohttpSender(timeout=timeout) + else: + raise ValueError( + f"HTTPTransport: backend must be 'urllib' or 'aiohttp', got {backend!r}" + ) self._batch: list[str] = [] + self._batch_bytes = 0 def write(self, formatted: str, record: LogRecord) -> None: - """Buffers `formatted` and triggers a `flush()` once `batch_size` is - reached.""" + """Buffers `formatted` and triggers a `flush()` once `batch_size` + records or `max_bytes` bytes are buffered.""" self._batch.append(formatted) - if len(self._batch) >= self.batch_size: + self._batch_bytes += len(formatted.encode("utf-8", errors="replace")) + if len(self._batch) >= self.batch_size or self._batch_bytes >= self.max_bytes: self.flush() def flush(self) -> None: - """Sends whatever is currently buffered via `sender`, clearing the - buffer first. No-op if nothing is buffered.""" + """Sends whatever is currently buffered via the sender, clearing the + buffer first. No-op if nothing is buffered. A failed send is logged, + not raised.""" if not self._batch: return batch, self._batch = self._batch, [] - self._sender(self.url, batch) + self._batch_bytes = 0 + try: + self._sender(self.url, batch) + except Exception: + _logger.exception( + "HTTPTransport: couldn't deliver %d log record(s) to %s — check the URL " + "is reachable and accepts POSTed NDJSON, or raise `timeout`; " + "those records were dropped", + len(batch), + self.url, + ) def close(self) -> None: - """Flushes any remaining buffered records.""" + """Flushes any remaining buffered records and releases the backend's + connection, if it holds one.""" self.flush() + closer = getattr(self._sender, "close", None) + if callable(closer): + closer() diff --git a/logquill/transports/nosql/dynamodb_transport.py b/logquill/transports/nosql/dynamodb_transport.py index bb2e42a..aa36cff 100644 --- a/logquill/transports/nosql/dynamodb_transport.py +++ b/logquill/transports/nosql/dynamodb_transport.py @@ -2,7 +2,7 @@ from typing import Any, ContextManager, Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/nosql/mongodb_transport.py b/logquill/transports/nosql/mongodb_transport.py index b2a56db..ff37835 100644 --- a/logquill/transports/nosql/mongodb_transport.py +++ b/logquill/transports/nosql/mongodb_transport.py @@ -2,7 +2,7 @@ from typing import Any, Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/nosql/redis_transport.py b/logquill/transports/nosql/redis_transport.py index e1efd75..4fcdb17 100644 --- a/logquill/transports/nosql/redis_transport.py +++ b/logquill/transports/nosql/redis_transport.py @@ -3,7 +3,7 @@ import json from typing import Any, Protocol, Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/queue/base_queue_transport.py b/logquill/transports/queue/base_queue_transport.py index 1bb400f..57aaf84 100644 --- a/logquill/transports/queue/base_queue_transport.py +++ b/logquill/transports/queue/base_queue_transport.py @@ -3,7 +3,7 @@ from abc import abstractmethod from typing import Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/queue/kafka_transport.py b/logquill/transports/queue/kafka_transport.py index e570795..eec8a48 100644 --- a/logquill/transports/queue/kafka_transport.py +++ b/logquill/transports/queue/kafka_transport.py @@ -3,7 +3,7 @@ import json from typing import Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.queue.base_queue_transport import BaseQueueTransport diff --git a/logquill/transports/queue/pubsub_transport.py b/logquill/transports/queue/pubsub_transport.py index 5ca92e5..d5ba608 100644 --- a/logquill/transports/queue/pubsub_transport.py +++ b/logquill/transports/queue/pubsub_transport.py @@ -3,7 +3,7 @@ import json from typing import Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.queue.base_queue_transport import BaseQueueTransport diff --git a/logquill/transports/queue/rabbitmq_transport.py b/logquill/transports/queue/rabbitmq_transport.py index a82a902..c7a1096 100644 --- a/logquill/transports/queue/rabbitmq_transport.py +++ b/logquill/transports/queue/rabbitmq_transport.py @@ -3,7 +3,7 @@ import json from typing import Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.queue.base_queue_transport import BaseQueueTransport diff --git a/logquill/transports/queue/sqs_transport.py b/logquill/transports/queue/sqs_transport.py index 0033133..9e341f5 100644 --- a/logquill/transports/queue/sqs_transport.py +++ b/logquill/transports/queue/sqs_transport.py @@ -3,7 +3,7 @@ import json from typing import Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.queue.base_queue_transport import BaseQueueTransport diff --git a/logquill/transports/sql/base_sql_transport.py b/logquill/transports/sql/base_sql_transport.py index 5dc5f5c..08b5a7b 100644 --- a/logquill/transports/sql/base_sql_transport.py +++ b/logquill/transports/sql/base_sql_transport.py @@ -4,7 +4,7 @@ from abc import abstractmethod from typing import Any, Sequence, TypedDict -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.records import LogRecord from logquill.transports.batching_transport import BatchingTransport diff --git a/logquill/transports/sql/mysql_transport.py b/logquill/transports/sql/mysql_transport.py index 013741d..f43d3fa 100644 --- a/logquill/transports/sql/mysql_transport.py +++ b/logquill/transports/sql/mysql_transport.py @@ -2,7 +2,7 @@ from typing import Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.transports.sql.base_sql_transport import BaseSQLTransport, SQLLogRow diff --git a/logquill/transports/sql/postgres_transport.py b/logquill/transports/sql/postgres_transport.py index 2d2efa5..80f3047 100644 --- a/logquill/transports/sql/postgres_transport.py +++ b/logquill/transports/sql/postgres_transport.py @@ -2,7 +2,7 @@ from typing import Protocol, Sequence, cast -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.transports.sql.base_sql_transport import BaseSQLTransport, SQLLogRow diff --git a/logquill/transports/sql/sqlite_transport.py b/logquill/transports/sql/sqlite_transport.py index 4c95f0b..521f2ce 100644 --- a/logquill/transports/sql/sqlite_transport.py +++ b/logquill/transports/sql/sqlite_transport.py @@ -3,7 +3,7 @@ import sqlite3 from typing import Iterable, Protocol, Sequence -from logquill.formatter import Formatter +from logquill.formatters import Formatter from logquill.transports.sql.base_sql_transport import BaseSQLTransport, SQLLogRow diff --git a/logquill/transports/transport.py b/logquill/transports/transport.py index b3b87f9..1427b9e 100644 --- a/logquill/transports/transport.py +++ b/logquill/transports/transport.py @@ -2,7 +2,7 @@ from abc import ABC, abstractmethod -from logquill.formatter import Formatter, JSONFormatter +from logquill.formatters import Formatter, JSONFormatter from logquill.records import LogRecord diff --git a/logquill/worker.py b/logquill/worker.py index 21dc645..411c30c 100644 --- a/logquill/worker.py +++ b/logquill/worker.py @@ -68,7 +68,7 @@ def __init__( self._pending = 0 self._closed = False self._cond = threading.Condition() - self._last_drop_warning = 0.0 + self._last_drop_warning: float | None = None self._thread = threading.Thread(target=self._run, name="logquill-worker", daemon=True) self._thread.start() @@ -105,7 +105,11 @@ def submit(self, item: WorkItem) -> None: def _warn_dropping(self) -> None: now = time.monotonic() - if now - self._last_drop_warning >= _DROP_WARNING_INTERVAL_SECONDS: + # `None` rather than 0.0: the monotonic clock's zero point is arbitrary + # (often boot time), so a fresh process could otherwise be inside the + # warning interval already and never warn about its first drops + last = self._last_drop_warning + if last is None or now - last >= _DROP_WARNING_INTERVAL_SECONDS: self._last_drop_warning = now _logger.warning( "AsyncWorker: queue full at max_queue_size=%d, dropping records " diff --git a/pyproject.toml b/pyproject.toml index a5ec3f2..90ab9b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ dependencies = [] [project.optional-dependencies] http = ["aiohttp>=3.9"] +apprise = ["apprise>=1.8"] postgres = ["psycopg2-binary>=2.9"] mysql = ["pymysql>=1.1"] mongodb = ["pymongo>=4.6"] @@ -96,6 +97,7 @@ packages = ["logquill"] include = [ "/logquill", "/tests", + "/benchmarks", "/README.md", "/CHANGELOG.md", "/CODE_OF_CONDUCT.md", @@ -128,6 +130,13 @@ files = ["logquill"] module = "logquill.transports.file_transport" warn_unused_ignores = false +# `apprise` and `aiohttp` are optional dependencies that a bare type-check +# environment (the pre-commit hook) doesn't install, so their imports have to +# resolve the same way whether or not they're present. +[[tool.mypy.overrides]] +module = ["apprise", "aiohttp"] +ignore_missing_imports = true + [tool.pytest.ini_options] testpaths = ["tests"] asyncio_mode = "auto" diff --git a/tests/adversarial.py b/tests/adversarial.py new file mode 100644 index 0000000..102a6e0 --- /dev/null +++ b/tests/adversarial.py @@ -0,0 +1,34 @@ +"""Hypothesis strategies for hostile `meta` payloads, shared by the property tests.""" + +from __future__ import annotations + +from hypothesis import strategies as st + +# Deliberately adversarial: deeply nested containers, unusual scalar types, +# and non-JSON-serializable values (a raw object, bytes). Circular +# references are exercised separately below, since hypothesis strategies +# can't easily generate them. +scalars = st.one_of( + st.none(), + st.booleans(), + st.integers(), + st.floats(allow_nan=True, allow_infinity=True), + st.text(), + st.binary(), + st.builds(object), +) + +meta_values = st.recursive( + scalars, + lambda children: st.one_of( + st.lists(children, max_size=5), + st.dictionaries(st.text(min_size=1, max_size=10), children, max_size=5), + ), + max_leaves=25, +) + +meta_dicts = st.dictionaries(st.text(min_size=1, max_size=10), meta_values, max_size=8) + +# Text that includes what `st.text()` normally leaves out (lone surrogates), +# for the formatters, which must cope with whatever string a caller passes. +hostile_text = st.text(alphabet=st.characters(blacklist_categories=()), max_size=40) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..be8e962 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from collections.abc import Iterator + +import pytest + +from logquill import toggle + + +@pytest.fixture(autouse=True) +def _reset_enable_disable_rules() -> Iterator[None]: + """`disable()`/`enable()` are process-wide; keep one test's rules from + silencing the next test's loggers.""" + toggle._reset() + yield + toggle._reset() diff --git a/tests/test_backpressure_burst.py b/tests/test_backpressure_burst.py new file mode 100644 index 0000000..512d05e --- /dev/null +++ b/tests/test_backpressure_burst.py @@ -0,0 +1,125 @@ +"""A burst of tens of thousands of log calls against a transport that is +stalled the whole time: each backpressure policy must do exactly what it +promises, and the queue must never grow past its bound.""" + +from __future__ import annotations + +import logging +import threading +import time + +import pytest + +from logquill import Logger +from logquill.records import LogRecord +from logquill.transports.transport import CollectingTransport + +BURST = 30_000 +QUEUE_LIMIT = 100 + + +class _Stalled(CollectingTransport): + def __init__(self) -> None: + super().__init__() + self.gate = threading.Event() + self.first_write_started = threading.Event() + + def write(self, formatted: str, record: LogRecord) -> None: + self.first_write_started.set() + self.gate.wait(timeout=30) + super().write(formatted, record) + + def delivered_indices(self) -> list[int]: + return [int(record["meta"]["i"]) for record in self.records] + + +def _stalled_logger(policy: str) -> tuple[Logger, _Stalled]: + transport = _Stalled() + logger = Logger( + "app.burst", + transports=[transport], + async_dispatch=True, + max_queue_size=QUEUE_LIMIT, + backpressure=policy, # type: ignore[arg-type] + flush_at_exit=False, + ) + logger.info("first", i=0) # the worker picks this up and stalls inside write() + assert transport.first_write_started.wait(timeout=5) + return logger, transport + + +def _burst(logger: Logger) -> None: + for i in range(1, BURST): + logger.info("burst", i=i) + + +def test_drop_oldest_keeps_the_newest_records_and_never_blocks_the_caller( + caplog: pytest.LogCaptureFixture, +) -> None: + logger, transport = _stalled_logger("drop_oldest") + + with caplog.at_level(logging.WARNING, logger="logquill"): + started = time.monotonic() + _burst(logger) # would hang here if the caller ever blocked on the stalled sink + elapsed = time.monotonic() - started + + assert logger._worker is not None + assert logger._worker.qsize == QUEUE_LIMIT + transport.gate.set() + logger.close(timeout=10) + + # the in-flight record, then exactly the newest QUEUE_LIMIT — in order + assert transport.delivered_indices() == [0, *range(BURST - QUEUE_LIMIT, BURST)] + assert elapsed < 20 + assert len([r for r in caplog.records if "queue full" in r.getMessage()]) == 1 + + +def test_drop_newest_keeps_the_oldest_records_and_never_blocks_the_caller() -> None: + logger, transport = _stalled_logger("drop_newest") + + _burst(logger) + + assert logger._worker is not None + assert logger._worker.qsize == QUEUE_LIMIT + transport.gate.set() + logger.close(timeout=10) + + assert transport.delivered_indices() == [0, *range(1, QUEUE_LIMIT + 1)] + + +def test_block_makes_the_caller_wait_and_then_delivers_every_record_in_order() -> None: + logger, transport = _stalled_logger("block") + burst = threading.Thread(target=_burst, args=(logger,), daemon=True) + + burst.start() + burst.join(timeout=0.5) + + assert burst.is_alive(), "the caller should be blocked on the full queue, not racing ahead" + assert logger._worker is not None + assert logger._worker.qsize == QUEUE_LIMIT + + transport.gate.set() + burst.join(timeout=30) + logger.close(timeout=30) + + assert not burst.is_alive() + assert transport.delivered_indices() == list(range(BURST)) + + +@pytest.mark.parametrize("policy", ["drop_oldest", "drop_newest"]) +def test_a_burst_below_the_limit_loses_nothing(policy: str) -> None: + transport = CollectingTransport() + logger = Logger( + "app.burst", + transports=[transport], + async_dispatch=True, + max_queue_size=QUEUE_LIMIT, + backpressure=policy, # type: ignore[arg-type] + flush_at_exit=False, + ) + + for i in range(QUEUE_LIMIT): + logger.info("burst", i=i) + logger.close(timeout=10) + + assert len(transport.records) == QUEUE_LIMIT diff --git a/tests/test_cli.py b/tests/test_cli.py index 5c44ba5..375f68b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -176,3 +176,23 @@ def test_tail_follow_picks_up_appended_records(tmp_path: Path) -> None: assert "second" in follow_out.getvalue() assert "first" not in follow_out.getvalue() + + +def test_tail_prints_a_stack_trace_on_its_own_lines(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + _write_lines( + log_path, + [ + { + "timestamp": "2026-01-01T00:00:00.000Z", + "level": "ERROR", + "logger": "app", + "message": "failed", + "meta": {"stack": "Traceback (most recent call last):\nValueError: nope\n"}, + } + ], + ) + + output, _warn, _exit_code = _tail(str(log_path)) + + assert output.splitlines()[1:] == ["Traceback (most recent call last):", "ValueError: nope"] diff --git a/tests/test_config.py b/tests/test_config.py index 33bf7c2..d65aaf3 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -177,3 +177,37 @@ def test_logger_from_env_custom_prefix(monkeypatch: pytest.MonkeyPatch) -> None: logger = logger_from_env(prefix="MYAPP_") assert logger.level == Level.ERROR + + +def test_transport_options_can_name_a_builtin_formatter() -> None: + from logquill import LogfmtFormatter, TextFormatter + + logger = load_config( + { + "transports": [ + {"type": "console", "options": {"formatter": "text"}}, + {"type": "console", "options": {"formatter": "logfmt"}}, + {"type": "console", "options": {"formatter": "json"}}, + ] + } + ) + + formatters = [transport.formatter for transport in logger.transports] + assert isinstance(formatters[0], TextFormatter) + assert isinstance(formatters[1], LogfmtFormatter) + assert type(formatters[2]).__name__ == "JSONFormatter" + + +def test_an_unknown_formatter_name_lists_the_valid_ones() -> None: + with pytest.raises(ValueError, match="Unknown formatter 'xml'.*json, logfmt, text"): + load_config({"transports": [{"type": "console", "options": {"formatter": "xml"}}]}) + + +def test_flush_at_exit_can_be_set_from_config() -> None: + from logquill import shutdown + + opted_out = load_config({"flush_at_exit": False}) + default = load_config({}) + + assert opted_out not in shutdown._loggers + assert default in shutdown._loggers diff --git a/tests/test_diagnose.py b/tests/test_diagnose.py new file mode 100644 index 0000000..34f0c0b --- /dev/null +++ b/tests/test_diagnose.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import logging + +import pytest + +from logquill import Logger, PIIRedactPlugin, Plugin, RedactPlugin +from logquill import exceptions as exceptions_module +from logquill.exceptions import format_exc_info +from logquill.transports.transport import CollectingTransport + +# Obviously fake values, kept in constants rather than assigned as literals to +# variables named `password`/`token`, which secret scanners flag on sight. +FAKE_PASSWORD = "placeholder-value-1" +FAKE_TOKEN = "placeholder-value-2" +FAKE_KEY = "placeholder-value-3" + + +def _fail_with_locals() -> None: + password = FAKE_PASSWORD + token = FAKE_TOKEN + email = "someone@example.com" + visible = "harmless-value" + print(password, token, email, visible) # noqa: T201 + raise ValueError("boom") + + +def _logger(*plugins: Plugin) -> tuple[Logger, CollectingTransport]: + sink = CollectingTransport() + return Logger("app", transports=[sink], plugins=list(plugins)), sink + + +def _stack_of(sink: CollectingTransport) -> str: + stack = sink.records[-1]["meta"]["stack"] + assert isinstance(stack, str) + return stack + + +def test_diagnose_is_off_by_default_and_shows_no_local_values() -> None: + logger, sink = _logger() + + try: + _fail_with_locals() + except ValueError as exc: + logger.error("failed", exc_info=exc) + + stack = _stack_of(sink) + assert "ValueError: boom" in stack + assert "harmless-value" not in stack + assert FAKE_PASSWORD not in stack + + +def test_diagnose_prints_local_values_under_each_frame() -> None: + logger, sink = _logger() + + try: + _fail_with_locals() + except ValueError: + logger.error("failed", diagnose=True) # implies exc_info=True + + stack = _stack_of(sink) + assert "visible = 'harmless-value'" in stack + assert "ValueError: boom" in stack + assert "diagnose" not in sink.records[0]["meta"] + + +def test_diagnose_output_contains_nothing_redact_plugin_would_mask() -> None: + logger, sink = _logger(RedactPlugin()) + + try: + _fail_with_locals() + except ValueError as exc: + logger.error("failed", exc_info=exc, diagnose=True) + + stack = _stack_of(sink) + assert FAKE_PASSWORD not in stack + assert FAKE_TOKEN not in stack + assert "password = ***" in stack + assert "token = ***" in stack + assert "harmless-value" in stack # only what the plugin masks is masked + + +def test_diagnose_output_contains_no_pii_pii_plugin_would_mask() -> None: + logger, sink = _logger(PIIRedactPlugin()) + + try: + _fail_with_locals() + except ValueError as exc: + logger.error("failed", exc_info=exc, diagnose=True) + + stack = _stack_of(sink) + assert "someone@example.com" not in stack + assert "email = '***'" in stack + + +def test_diagnose_redacts_chained_exceptions_too() -> None: + def inner() -> None: + api_key = FAKE_KEY + raise KeyError(len(api_key)) + + def outer() -> None: + try: + inner() + except KeyError as exc: + raise RuntimeError("wrapped") from exc + + logger, sink = _logger(RedactPlugin()) + try: + outer() + except RuntimeError as exc: + logger.error("failed", exc_info=exc, diagnose=True) + + stack = _stack_of(sink) + assert FAKE_KEY not in stack + assert "api_key = ***" in stack + assert "RuntimeError: wrapped" in stack and "KeyError" in stack + + +def test_a_redaction_hook_that_raises_fails_closed() -> None: + class Broken(Plugin): + def redact_local(self, name: str, text: str) -> str: + raise RuntimeError("bug in plugin") + + logger, sink = _logger(Broken()) + + try: + _fail_with_locals() + except ValueError: + logger.error("failed", diagnose=True) + + stack = _stack_of(sink) + assert FAKE_PASSWORD not in stack + assert "harmless-value" not in stack + assert "" in stack + + +def test_a_local_with_a_raising_repr_falls_back_to_a_plain_traceback() -> None: + class BadRepr: + def __repr__(self) -> str: + raise RuntimeError("no repr") + + def fail() -> None: + bad = BadRepr() # noqa: F841 + raise ValueError("boom") + + logger, sink = _logger() + try: + fail() + except ValueError: + logger.error("failed", diagnose=True) + + stack = _stack_of(sink) + assert "ValueError: boom" in stack + assert "BadRepr" not in stack + + +def test_long_local_reprs_are_truncated() -> None: + def fail() -> None: + blob = "x" * 5000 # noqa: F841 + raise ValueError("boom") + + try: + fail() + except ValueError as exc: + stack = format_exc_info(exc, diagnose=True) + + assert stack is not None + assert "x" * 5000 not in stack + assert "..." in stack + + +def test_diagnose_without_a_current_exception_adds_no_stack() -> None: + logger, sink = _logger() + + logger.error("nothing is being handled", diagnose=True) + + assert "stack" not in sink.records[0]["meta"] + + +def test_diagnose_warns_once_about_leaking_sensitive_data( + caplog: pytest.LogCaptureFixture, +) -> None: + exceptions_module._diagnose_warned = False + logger, _ = _logger() + + with caplog.at_level(logging.WARNING, logger="logquill"): + for _ in range(3): + try: + _fail_with_locals() + except ValueError: + logger.error("failed", diagnose=True) + + warnings = [r for r in caplog.records if "leak sensitive data" in r.getMessage()] + assert len(warnings) == 1 diff --git a/tests/test_formatter_properties.py b/tests/test_formatter_properties.py new file mode 100644 index 0000000..b17213e --- /dev/null +++ b/tests/test_formatter_properties.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import json +import re +from typing import Any + +from adversarial import hostile_text, meta_dicts +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from logquill import JSONFormatter, LogfmtFormatter, TextFormatter, parse_logfmt +from logquill.records import LogRecord + +_settings = settings(max_examples=150, suppress_health_check=[HealthCheck.too_slow]) + +_LINE_BREAKS = re.compile(r"[\n\r\x0b\x0c\x1c-\x1e\x85

]") + + +def _record(message: str, meta: dict[str, Any]) -> LogRecord: + return LogRecord( + timestamp="2026-01-01T00:00:00.000Z", + level="INFO", + logger="app.test", + message=message, + meta=meta, + ) + + +@_settings +@given(message=hostile_text, meta=meta_dicts) +def test_text_formatter_never_raises_and_returns_a_string( + message: str, meta: dict[str, Any] +) -> None: + line = TextFormatter().format(_record(message, meta)) + + assert isinstance(line, str) + assert line.startswith("2026-01-01T00:00:00.000Z INFO app.test: ") + + +@_settings +@given(message=hostile_text, meta=meta_dicts) +def test_logfmt_formatter_never_raises_and_always_emits_exactly_one_line( + message: str, meta: dict[str, Any] +) -> None: + line = LogfmtFormatter().format(_record(message, meta)) + + assert isinstance(line, str) + assert not _LINE_BREAKS.search(line) + + +@_settings +@given(message=hostile_text) +def test_logfmt_round_trips_any_message(message: str) -> None: + fields = parse_logfmt(LogfmtFormatter().format(_record(message, {}))) + + assert fields["message"] == message + assert fields["level"] == "INFO" + assert fields["logger"] == "app.test" + + +_safe_keys = st.from_regex(r"[a-z][a-z0-9_]{0,9}", fullmatch=True).filter( + lambda key: key not in {"timestamp", "level", "logger", "message"} +) + + +@_settings +@given(meta=st.dictionaries(_safe_keys, hostile_text, max_size=6)) +def test_logfmt_round_trips_string_meta_values(meta: dict[str, str]) -> None: + fields = parse_logfmt(LogfmtFormatter().format(_record("m", dict(meta)))) + + assert {key: fields[key] for key in meta} == meta + + +@_settings +@given(meta=meta_dicts) +def test_json_formatter_emits_parseable_json_for_any_non_circular_meta( + meta: dict[str, Any], +) -> None: + parsed = json.loads(JSONFormatter().format(_record("m", meta))) + + assert parsed["message"] == "m" + assert isinstance(parsed["meta"], dict) diff --git a/tests/test_formatters.py b/tests/test_formatters.py new file mode 100644 index 0000000..540b58b --- /dev/null +++ b/tests/test_formatters.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import io + +from logquill import ( + ConsoleTransport, + LogfmtFormatter, + Logger, + TextFormatter, + parse_logfmt, +) +from logquill.levels import Level +from logquill.records import LogRecord, create_record + + +def _record(message: str = "user signed up", **meta: object) -> LogRecord: + record = create_record(level=Level.INFO, logger="app.api", message=message, meta=dict(meta)) + record["timestamp"] = "2026-01-01T00:00:00.000Z" + return record + + +def test_old_import_path_still_works() -> None: + from logquill.formatter import Formatter, JSONFormatter + from logquill.formatters import Formatter as NewFormatter + from logquill.formatters import JSONFormatter as NewJSONFormatter + + assert Formatter is NewFormatter + assert JSONFormatter is NewJSONFormatter + + +# --- text ----------------------------------------------------------------- + + +def test_text_formatter_renders_one_readable_line() -> None: + line = TextFormatter().format(_record(user_id=42)) + + assert line == '2026-01-01T00:00:00.000Z INFO app.api: user signed up {"user_id":42}' + + +def test_text_formatter_omits_meta_when_empty() -> None: + assert ( + TextFormatter().format(_record()) + == "2026-01-01T00:00:00.000Z INFO app.api: user signed up" + ) + + +def test_text_formatter_prints_a_stack_on_following_lines() -> None: + stack = 'Traceback (most recent call last):\n File "x.py", line 1\nValueError: nope\n' + + line = TextFormatter().format(_record("failed", stack=stack, attempt=2)) + + first, *rest = line.split("\n") + assert first.endswith('failed {"attempt":2}') + assert rest == stack.rstrip("\n").split("\n") + + +def test_text_formatter_survives_circular_and_unprintable_meta() -> None: + class Unprintable: + def __repr__(self) -> str: + raise RuntimeError("no repr") + + __str__ = __repr__ + + loop: dict[str, object] = {} + loop["self"] = loop + + line = TextFormatter().format(_record(loop=loop, bad=Unprintable())) + + assert line.startswith("2026-01-01T00:00:00.000Z INFO app.api: user signed up ") + assert "" in line + + +def test_console_transport_with_text_formatter() -> None: + out = io.StringIO() + logger = Logger( + "app", + transports=[ConsoleTransport(formatter=TextFormatter(), colorize=False, stdout=out)], + ) + + logger.info("hello", n=1) + + assert out.getvalue().endswith('app: hello {"n":1}\n') + + +# --- logfmt --------------------------------------------------------------- + + +def test_logfmt_formatter_emits_core_fields_then_meta() -> None: + line = LogfmtFormatter().format(_record(user_id=42, plan="pro")) + + assert line == ( + "timestamp=2026-01-01T00:00:00.000Z level=INFO logger=app.api " + 'message="user signed up" user_id=42 plan=pro' + ) + + +def test_logfmt_quotes_and_escapes_values_so_the_line_stays_single_line() -> None: + line = LogfmtFormatter().format(_record(note='a "quoted"\nline\twith = sign', empty="")) + + assert "\n" not in line and "\t" not in line + assert r'note="a \"quoted\"\nline\twith = sign"' in line + assert 'empty=""' in line + + +def test_logfmt_flattens_nested_dicts_and_encodes_scalars() -> None: + line = LogfmtFormatter().format( + _record(http={"status": 200, "ok": True, "body": None}, tags=["a", "b"]) + ) + + assert "http.status=200 http.ok=true http.body=null" in line + assert 'tags="[\\"a\\",\\"b\\"]"' in line + + +def test_logfmt_prefixes_meta_keys_that_shadow_core_fields() -> None: + record = _record() + record["meta"] = {"level": "custom", "message": "also"} + + line = LogfmtFormatter().format(record) + + fields = parse_logfmt(line) + assert fields["level"] == "INFO" + assert fields["message"] == "user signed up" + assert fields["meta.level"] == "custom" + assert fields["meta.message"] == "also" + + +def test_logfmt_sanitizes_awkward_keys() -> None: + line = LogfmtFormatter().format(_record(**{"has space": 1, "a=b": 2, "": 3})) + + assert "has_space=1" in line and "a_b=2" in line and " _=3" in line + + +def test_logfmt_deep_nesting_falls_back_to_a_json_value() -> None: + deep: dict[str, object] = {} + node = deep + for _ in range(20): + child: dict[str, object] = {} + node["n"] = child + node = child + node["leaf"] = 1 + + line = LogfmtFormatter().format(_record(deep=deep)) + + assert "\n" not in line + assert line.count("=") < 15 # bounded, not one pair per level + + +def test_logfmt_round_trips_through_parse_logfmt() -> None: + record = _record(note='tricky "value" \\ with\nnewline', n=7, empty="") + + fields = parse_logfmt(LogfmtFormatter().format(record)) + + assert fields["note"] == 'tricky "value" \\ with\nnewline' + assert fields["n"] == "7" + assert fields["empty"] == "" + assert fields["message"] == "user signed up" diff --git a/tests/test_logger.py b/tests/test_logger.py index 3641d71..e50d1c5 100644 --- a/tests/test_logger.py +++ b/tests/test_logger.py @@ -60,3 +60,14 @@ def test_a_meta_key_named_message_does_not_collide_with_the_positional_arg() -> assert record is not None assert record["message"] == "hello" assert record["meta"] == {"message": "not the real message"} + + +def test_a_malformed_exc_info_is_ignored_instead_of_crashing_the_caller() -> None: + logger = Logger("app.test") + + for bad in ([None, None, []], (1,), "boom", 42, object()): + record = logger.error("still logs", exc_info=bad) + + assert record is not None + assert "stack" not in record["meta"] + assert "exc_info" not in record["meta"] diff --git a/tests/test_opt.py b/tests/test_opt.py new file mode 100644 index 0000000..41520d0 --- /dev/null +++ b/tests/test_opt.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import pytest + +from logquill import Level, Logger +from logquill.transports.transport import CollectingTransport + + +def _logger(level: str = "DEBUG") -> tuple[Logger, CollectingTransport]: + sink = CollectingTransport() + return Logger("app.test", level=level, transports=[sink]), sink + + +def test_lazy_values_are_not_evaluated_when_the_level_filters_the_call_out() -> None: + logger, sink = _logger(level="INFO") + calls: list[int] = [] + + result = logger.opt(lazy=True).debug("state", dump=lambda: calls.append(1)) + + assert result is None + assert calls == [] + assert sink.records == [] + + +def test_lazy_values_are_evaluated_when_the_record_is_emitted() -> None: + logger, sink = _logger() + + logger.opt(lazy=True).debug("state", total=lambda: 2 + 2, plain=5) + + assert sink.records[0]["meta"] == {"total": 4, "plain": 5} + + +def test_lazy_values_are_not_evaluated_for_a_disabled_logger() -> None: + import logquill + + logger, _ = _logger() + logquill.disable("app") + calls: list[int] = [] + + logger.opt(lazy=True).info("x", v=lambda: calls.append(1)) + + assert calls == [] + + +def test_a_raising_lazy_value_becomes_a_placeholder_instead_of_crashing_the_caller() -> None: + logger, sink = _logger() + + logger.opt(lazy=True).info("x", bad=lambda: 1 / 0, good=lambda: "ok") + + meta = sink.records[0]["meta"] + assert meta["bad"] == "" + assert meta["good"] == "ok" + + +def test_callables_are_left_alone_without_lazy() -> None: + logger, sink = _logger() + + def fn() -> None: ... + + logger.opt().info("x", callback=fn) + + assert sink.records[0]["meta"]["callback"] is fn + + +def test_async_dispatch_evaluates_lazy_values_on_the_calling_thread() -> None: + import threading + + sink = CollectingTransport() + logger = Logger("app", level="DEBUG", transports=[sink], async_dispatch=True) + seen: list[int] = [] + + logger.opt(lazy=True).info("x", v=lambda: seen.append(threading.get_ident())) + logger.close() + + assert seen == [threading.get_ident()] + + +def test_depth_zero_reports_the_direct_caller() -> None: + logger, sink = _logger() + + logger.opt(depth=0).info("here") + + caller = sink.records[0]["meta"]["caller"] + assert caller["function"] == "test_depth_zero_reports_the_direct_caller" + assert caller["module"] == __name__ + assert caller["file"].endswith("test_opt.py") + assert isinstance(caller["line"], int) + + +def test_depth_one_skips_a_wrapper_and_reports_its_caller() -> None: + logger, sink = _logger() + + def wrapper(message: str) -> None: + logger.opt(depth=1).info(message) + + def business_logic() -> None: + wrapper("via wrapper") + + business_logic() + + assert sink.records[0]["meta"]["caller"]["function"] == "business_logic" + + +def test_depth_works_for_every_method_family() -> None: + logger, sink = _logger(level="TRACE") + view = logger.opt(depth=0) + + for method in ( + view.trace, + view.debug, + view.info, + view.warn, + view.error, + view.fatal, + view.thought, + view.action, + view.observation, + view.decision, + ): + method("m") + + functions = {record["meta"]["caller"]["function"] for record in sink.records} + assert functions == {"test_depth_works_for_every_method_family"} + assert [r["level"] for r in sink.records][:6] == [level.name for level in Level] + assert sink.records[6]["meta"]["kind"] == "thought" + + +def test_depth_beyond_the_stack_omits_caller_instead_of_raising() -> None: + logger, sink = _logger() + + logger.opt(depth=10_000).info("x") + + assert "caller" not in sink.records[0]["meta"] + + +def test_an_explicit_caller_in_meta_wins() -> None: + logger, sink = _logger() + + logger.opt(depth=0).info("x", caller="mine") + + assert sink.records[0]["meta"]["caller"] == "mine" + + +def test_no_caller_is_recorded_without_depth() -> None: + logger, sink = _logger() + + logger.info("x") + logger.opt(lazy=True).info("y") + + assert all("caller" not in record["meta"] for record in sink.records) + + +def test_negative_depth_is_rejected() -> None: + logger, _ = _logger() + + with pytest.raises(ValueError, match="depth"): + logger.opt(depth=-1) diff --git a/tests/test_parsing.py b/tests/test_parsing.py new file mode 100644 index 0000000..aa6dff4 --- /dev/null +++ b/tests/test_parsing.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import io +import re +from pathlib import Path + +import pytest + +from logquill import ( + TEXT_LOG_CASTS, + TEXT_LOG_PATTERN, + Logger, + TextFormatter, + parse, + parse_logfmt, +) + +LEGACY = r"(?P\S+ \S+) \[(?P[A-Z]+)\] (?P\d+) (?P.*)" + + +def _write(path: Path, *lines: str) -> Path: + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return path + + +def test_parse_extracts_named_groups_from_a_legacy_log_file(tmp_path: Path) -> None: + path = _write( + tmp_path / "legacy.log", + "2026-01-01 10:00:00 [INFO] 200 started", + "garbage that matches nothing", + "2026-01-01 10:00:05 [ERROR] 503 upstream down", + ) + + entries = list(parse(path, LEGACY, cast={"code": int})) + + assert entries == [ + {"when": "2026-01-01 10:00:00", "level": "INFO", "code": 200, "message": "started"}, + {"when": "2026-01-01 10:00:05", "level": "ERROR", "code": 503, "message": "upstream down"}, + ] + + +def test_parse_accepts_a_string_path_a_compiled_pattern_and_any_line_iterable( + tmp_path: Path, +) -> None: + path = _write(tmp_path / "a.log", "2026-01-01 10:00:00 [INFO] 200 ok") + + from_str = list(parse(str(path), re.compile(LEGACY))) + from_stream = list(parse(io.StringIO("2026-01-01 10:00:00 [INFO] 200 ok\n"), LEGACY)) + from_list = list(parse(["2026-01-01 10:00:00 [INFO] 200 ok\r\n"], LEGACY)) + + assert from_str == from_stream == from_list + assert from_str[0]["message"] == "ok" + + +def test_parse_leaves_optional_groups_that_did_not_match_as_none() -> None: + entries = list(parse(["a b", "a"], r"(?P\w)(?: (?P\w))?$", cast={"second": str})) + + assert entries == [{"first": "a", "second": "b"}, {"first": "a", "second": None}] + + +def test_parse_is_lazy_and_streams(tmp_path: Path) -> None: + lines = iter(["2026-01-01 10:00:00 [INFO] 1 a", "2026-01-01 10:00:00 [INFO] 2 b"]) + consumed = 0 + + def source() -> object: + nonlocal consumed + for line in lines: + consumed += 1 + yield line + + iterator = parse(source(), LEGACY) # type: ignore[arg-type] + assert consumed == 0 + next(iterator) + assert consumed == 1 + + +def test_parse_rejects_a_pattern_without_named_groups() -> None: + with pytest.raises(ValueError, match="no named groups"): + parse(["x"], r"\d+") + + +def test_parse_rejects_a_cast_for_an_unknown_group() -> None: + with pytest.raises(ValueError, match="not named groups"): + parse(["x"], r"(?Px)", cast={"b": int}) + + +def test_parse_reports_the_line_number_when_a_cast_fails() -> None: + lines = ["2026-01-01 10:00:00 [INFO] 200 fine", "2026-01-01 10:00:00 [INFO] 99999999 fine"] + + def strict(value: str) -> int: + if len(value) > 3: + raise ValueError("too long") + return int(value) + + with pytest.raises(ValueError, match=r"line 2: cast for 'code' failed"): + list(parse(lines, LEGACY, cast={"code": strict})) + + +def test_parse_reads_back_what_text_formatter_wrote() -> None: + logger = Logger("app.api") + record = logger.info("user signed up", user_id=42, note="two words") + assert record is not None + line = TextFormatter().format(record) + + (entry,) = parse([line], TEXT_LOG_PATTERN, cast=TEXT_LOG_CASTS) + + assert entry["level"] == "INFO" + assert entry["logger"] == "app.api" + assert entry["message"] == "user signed up" + assert entry["meta"] == {"user_id": 42, "note": "two words"} + + +def test_parse_logfmt_handles_bare_keys_duplicates_and_escapes() -> None: + fields = parse_logfmt(r'flag a=1 a=2 msg="say \"hi\"\n" u="é" plain=x=y') + + assert fields["flag"] == "" + assert fields["a"] == "2" + assert fields["msg"] == 'say "hi"\n' + assert fields["u"] == "é" + assert fields["plain"] == "x=y" diff --git a/tests/test_plugin_pipeline_properties.py b/tests/test_plugin_pipeline_properties.py index 3da43a1..8ab20b6 100644 --- a/tests/test_plugin_pipeline_properties.py +++ b/tests/test_plugin_pipeline_properties.py @@ -2,8 +2,8 @@ from typing import Any +from adversarial import meta_dicts from hypothesis import HealthCheck, given, settings -from hypothesis import strategies as st from logquill.logger import Logger from logquill.plugins.context_plugin import ContextPlugin @@ -12,31 +12,6 @@ from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin from logquill.transports.transport import CollectingTransport -# Deliberately adversarial: deeply nested containers, unusual scalar types, -# and non-JSON-serializable values (a raw object, bytes). Circular -# references are exercised separately below, since hypothesis strategies -# can't easily generate them. -_scalars = st.one_of( - st.none(), - st.booleans(), - st.integers(), - st.floats(allow_nan=True, allow_infinity=True), - st.text(), - st.binary(), - st.builds(object), -) - -_meta_values = st.recursive( - _scalars, - lambda children: st.one_of( - st.lists(children, max_size=5), - st.dictionaries(st.text(min_size=1, max_size=10), children, max_size=5), - ), - max_leaves=25, -) - -_meta_dicts = st.dictionaries(st.text(min_size=1, max_size=10), _meta_values, max_size=8) - def _build_logger() -> tuple[Logger, CollectingTransport]: sink = CollectingTransport() @@ -54,7 +29,7 @@ def _build_logger() -> tuple[Logger, CollectingTransport]: @settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow]) -@given(meta=_meta_dicts) +@given(meta=meta_dicts) def test_pipeline_never_crashes_on_adversarial_meta(meta: dict[str, Any]) -> None: logger, _sink = _build_logger() diff --git a/tests/test_plugins/test_apprise_alert_plugin.py b/tests/test_plugins/test_apprise_alert_plugin.py new file mode 100644 index 0000000..93e1651 --- /dev/null +++ b/tests/test_plugins/test_apprise_alert_plugin.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import sys +import threading +import types +from typing import Any + +import pytest + +from logquill import AppriseAlertPlugin, Level, Logger +from logquill.records import LogRecord, create_record + + +class FakeApprise: + """Stands in for `apprise.Apprise`, recording what would have been sent.""" + + def __init__(self, delivers: bool = True) -> None: + self.delivers = delivers + self.urls: list[str] = [] + self.sent: list[dict[str, Any]] = [] + self.sent_event = threading.Event() + + def add(self, url: str) -> bool: + if url.startswith("bad://"): + return False + self.urls.append(url) + return True + + def notify(self, *, body: str, title: str, notify_type: str) -> bool: + self.sent.append({"body": body, "title": title, "notify_type": notify_type}) + self.sent_event.set() + return self.delivers + + +def _record(level: Level = Level.ERROR, message: str = "boom") -> LogRecord: + return create_record(level=level, logger="app.api", message=message, meta={"secret": "x"}) + + +def test_send_alert_notifies_with_title_body_and_failure_type() -> None: + client = FakeApprise() + plugin = AppriseAlertPlugin("json://x", apprise_client=client) + + plugin.send_alert(_record(), 1) + + assert client.sent == [ + {"body": "boom", "title": "[ERROR] app.api", "notify_type": "failure"}, + ] + + +def test_send_alert_includes_the_occurrence_count_but_not_meta() -> None: + client = FakeApprise() + plugin = AppriseAlertPlugin("json://x", apprise_client=client, title="Prod alert") + + plugin.send_alert(_record(), 7) + + assert client.sent[0]["body"] == "boom (x7)" + assert client.sent[0]["title"] == "Prod alert" + assert "secret" not in str(client.sent[0]) + + +def test_warn_maps_to_warning_type() -> None: + client = FakeApprise() + plugin = AppriseAlertPlugin("json://x", apprise_client=client, threshold="WARN") + + plugin.send_alert(_record(Level.WARN), 1) + + assert client.sent[0]["notify_type"] == "warning" + + +def test_a_failed_delivery_raises_an_actionable_error() -> None: + plugin = AppriseAlertPlugin("json://x", apprise_client=FakeApprise(delivers=False)) + + with pytest.raises(RuntimeError, match="failed to send"): + plugin.send_alert(_record(), 1) + + +def test_an_error_record_alerts_from_a_background_thread_and_failures_reach_on_error() -> None: + client = FakeApprise(delivers=False) + errors: list[Exception] = [] + seen = threading.Event() + + class Recording(AppriseAlertPlugin): + def on_error(self, exc: Exception, record: LogRecord) -> None: + errors.append(exc) + seen.set() + + logger = Logger("app", plugins=[Recording("json://x", apprise_client=client)]) + + logger.error("boom") + + assert seen.wait(timeout=5) + assert client.sent[0]["body"] == "boom" + assert isinstance(errors[0], RuntimeError) + + +def _install_fake_apprise(monkeypatch: pytest.MonkeyPatch, client: FakeApprise) -> None: + module = types.ModuleType("apprise") + module.Apprise = lambda: client # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "apprise", module) + + +def test_urls_are_registered_with_apprise_at_construction( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = FakeApprise() + _install_fake_apprise(monkeypatch, client) + + AppriseAlertPlugin(["discord://a/b", "ntfy://topic"]) + AppriseAlertPlugin("json://single") + + assert client.urls == ["discord://a/b", "ntfy://topic", "json://single"] + + +def test_an_unrecognized_url_fails_at_construction_not_on_the_first_alert( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _install_fake_apprise(monkeypatch, FakeApprise()) + + with pytest.raises(ValueError, match="doesn't recognize the service URL 'bad://nope'"): + AppriseAlertPlugin("bad://nope") + + +def test_a_missing_apprise_dependency_gives_an_install_hint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setitem(sys.modules, "apprise", None) # makes `import apprise` raise ImportError + + with pytest.raises(ImportError, match=r"pip install logquill\[apprise\]"): + AppriseAlertPlugin("json://x") + + +def test_delivers_through_the_real_apprise_library() -> None: + pytest.importorskip("apprise") + import json + from http.server import BaseHTTPRequestHandler, HTTPServer + + received: list[dict[str, Any]] = [] + arrived = threading.Event() + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers["Content-Length"]) + received.append(json.loads(self.rfile.read(length))) + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + arrived.set() + + def log_message(self, format: str, *args: Any) -> None: + pass + + httpd = HTTPServer(("127.0.0.1", 0), Handler) + threading.Thread( + target=httpd.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ).start() + try: + plugin = AppriseAlertPlugin(f"json://127.0.0.1:{httpd.server_port}/hook") + plugin.send_alert(_record(message="disk full"), 1) + finally: + httpd.shutdown() + httpd.server_close() + + assert arrived.is_set() + assert received[0]["message"] == "disk full" + assert received[0]["title"] == "[ERROR] app.api" + assert received[0]["type"] == "failure" diff --git a/tests/test_shutdown.py b/tests/test_shutdown.py new file mode 100644 index 0000000..9fb5ae7 --- /dev/null +++ b/tests/test_shutdown.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import subprocess +import sys +import textwrap +from pathlib import Path + +from logquill import Logger, shutdown +from logquill.transports.transport import CollectingTransport + +_SCRIPT = """ +import sys +from logquill import {imports} + +log = Logger("app", transports=[FileTransport(sys.argv[1])], {options}) +for i in range(200): + log.info("record", i=i) +# no flush(), no close(): the exit hook has to do it +""" + + +def _run(tmp_path: Path, *, imports: str, options: str) -> list[str]: + path = tmp_path / "out.log" + script = textwrap.dedent(_SCRIPT).format(imports=imports, options=options) + subprocess.run([sys.executable, "-c", script, str(path)], check=True, timeout=60) + return path.read_text().splitlines() if path.exists() else [] + + +def test_exit_hook_flushes_an_async_queue_the_script_never_closed(tmp_path: Path) -> None: + lines = _run(tmp_path, imports="Logger, FileTransport", options="async_dispatch=True") + + assert len(lines) == 200 + + +def test_exit_hook_sends_a_batching_transports_unsent_batch(tmp_path: Path) -> None: + path = tmp_path / "sent.txt" + script = textwrap.dedent( + """ + import sys + from logquill import Logger, HTTPTransport + + def sender(url, batch): + with open(sys.argv[1], "a") as f: + f.write("\\n".join(batch) + "\\n") + + transport = HTTPTransport("http://unused", batch_size=1000, sender=sender) + log = Logger("app", transports=[transport]) + log.info("one") + log.info("two") + """ + ) + + subprocess.run([sys.executable, "-c", script, str(path)], check=True, timeout=60) + + assert len(path.read_text().splitlines()) == 2 + + +def test_flush_at_exit_false_leaves_the_logger_unregistered() -> None: + opted_out = Logger("app", flush_at_exit=False) + default = Logger("app") + + assert opted_out not in shutdown._loggers + assert default in shutdown._loggers + + +def test_shutdown_drains_the_queue_and_closes_each_shared_transport_once() -> None: + class CountingTransport(CollectingTransport): + close_calls = 0 + + def close(self) -> None: + type(self).close_calls += 1 + super().close() + + sink = CountingTransport() + parent = Logger("app", transports=[sink], async_dispatch=True) + child = parent.child("child") + parent.info("a") + child.info("b") + + shutdown.shutdown(timeout=2.0) + shutdown.shutdown(timeout=2.0) # safe to call twice + + assert len(sink.records) == 2 + assert CountingTransport.close_calls == 1 + + +def test_an_explicitly_closed_logger_is_not_closed_again_by_the_hook() -> None: + class CountingTransport(CollectingTransport): + close_calls = 0 + + def close(self) -> None: + type(self).close_calls += 1 + super().close() + + sink = CountingTransport() + logger = Logger("app", transports=[sink]) + child = logger.child("child") # shares the transport, registers separately + + logger.close() + shutdown.shutdown() + + assert child is not None + assert CountingTransport.close_calls == 1 + + +def test_a_transport_that_raises_on_close_does_not_stop_the_others() -> None: + class Exploding(CollectingTransport): + def close(self) -> None: + raise RuntimeError("cannot close") + + good = CollectingTransport() + logger = Logger("app", transports=[Exploding(), good]) + + shutdown.shutdown() + + assert good.closed is True + assert logger is not None + + +def test_registering_does_not_keep_a_logger_alive() -> None: + import gc + import weakref + + logger = Logger("app") + ref = weakref.ref(logger) + + del logger + gc.collect() + + assert ref() is None diff --git a/tests/test_toggle.py b/tests/test_toggle.py new file mode 100644 index 0000000..3aa6e3d --- /dev/null +++ b/tests/test_toggle.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +import logquill +from logquill import Logger +from logquill.transports.transport import CollectingTransport + + +def _logger(name: str) -> tuple[Logger, CollectingTransport]: + sink = CollectingTransport() + return Logger(name, transports=[sink]), sink + + +def test_everything_is_enabled_by_default() -> None: + logger, sink = _logger("mylib") + + logger.info("hello") + + assert len(sink.records) == 1 + assert logquill.is_enabled("mylib") + + +def test_disable_silences_a_logger_and_everything_nested_under_it() -> None: + parent, parent_sink = _logger("mylib") + child = parent.child("http") + sibling, sibling_sink = _logger("mylib2") + + logquill.disable("mylib") + + assert parent.info("x") is None + assert child.info("x") is None + assert sibling.info("x") is not None + assert parent_sink.records == [] + assert len(sibling_sink.records) == 1 + + +def test_enable_turns_a_disabled_library_back_on() -> None: + logger, sink = _logger("mylib") + logquill.disable("mylib") + logquill.enable("mylib") + + logger.info("visible again") + + assert len(sink.records) == 1 + + +def test_the_most_specific_rule_wins() -> None: + quiet, _ = _logger("mylib.internal") + loud, _ = _logger("mylib.http") + logquill.disable("mylib") + logquill.enable("mylib.http") + + assert quiet.info("x") is None + assert loud.info("x") is not None + + +def test_disable_with_no_name_silences_everything_and_enable_can_carve_out() -> None: + logquill.disable() + other, _ = _logger("other") + mine, _ = _logger("app") + logquill.enable("app") + + assert other.info("x") is None + assert mine.info("x") is not None + + +def test_disabled_spans_and_stdlib_bridge_records_are_dropped_too() -> None: + import logging + + logger, sink = _logger("mylib") + logquill.disable("mylib") + handler = logquill.LogQuillHandler(logger) + stdlib = logging.getLogger("toggle-test") + stdlib.addHandler(handler) + stdlib.propagate = False + try: + with logger.span("work"): + pass + stdlib.warning("nope") + finally: + stdlib.removeHandler(handler) + + assert sink.records == [] + + +def test_disable_rejects_a_non_string_name() -> None: + import pytest + + with pytest.raises(TypeError, match="logger name"): + logquill.disable(123) # type: ignore[arg-type] diff --git a/tests/test_transport_properties.py b/tests/test_transport_properties.py new file mode 100644 index 0000000..3fb7529 --- /dev/null +++ b/tests/test_transport_properties.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import io +import tempfile +from pathlib import Path +from typing import Any, Sequence + +from adversarial import meta_dicts +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from logquill import ( + BatchingTransport, + ConsoleTransport, + FileTransport, + HTTPTransport, + LogfmtFormatter, + Logger, + LogRecord, + TextFormatter, +) + +_settings = settings(max_examples=100, suppress_health_check=[HealthCheck.too_slow]) + + +@_settings +@given(meta=meta_dicts) +def test_console_transport_never_crashes_the_caller(meta: dict[str, Any]) -> None: + for formatter in (None, TextFormatter(), LogfmtFormatter()): + out, err = io.StringIO(), io.StringIO() + transport = ConsoleTransport(formatter=formatter, stdout=out, stderr=err) + logger = Logger("app", transports=[transport]) + + logger.info("adversarial", **meta) + logger.error("adversarial", **meta) + + +@_settings +@given(meta=meta_dicts) +def test_file_transport_never_crashes_the_caller_and_keeps_one_record_per_line( + meta: dict[str, Any], +) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "app.log" + transport = FileTransport(path) + logger = Logger("app", transports=[transport]) + + logger.info("adversarial", **meta) + logger.info("after") + transport.close() + + lines = path.read_text(encoding="utf-8").splitlines() + assert lines[-1].endswith('"message":"after","meta":{}}') + + +@_settings +@given(meta=meta_dicts) +def test_file_transport_with_a_text_formatter_survives_hostile_meta( + meta: dict[str, Any], +) -> None: + with tempfile.TemporaryDirectory() as directory: + transport = FileTransport(Path(directory) / "app.log", formatter=LogfmtFormatter()) + logger = Logger("app", transports=[transport]) + + logger.info("adversarial", **meta) + logger.info("after") + transport.close() + + +@_settings +@given(metas=st.lists(meta_dicts, max_size=30)) +def test_http_transport_buffer_stays_bounded_and_every_record_is_delivered_once( + metas: list[dict[str, Any]], +) -> None: + sent: list[str] = [] + + def sender(url: str, batch: Sequence[str]) -> None: + sent.extend(batch) + + transport = HTTPTransport("http://unused", batch_size=5, max_bytes=2_000, sender=sender) + logger = Logger("app", transports=[transport]) + + for meta in metas: + logger.info("adversarial", **meta) + assert len(transport._batch) < 5 + assert transport._batch_bytes < 2_000 + + transport.close() + assert len(sent) == len(metas) + + +class _Recording(BatchingTransport[LogRecord]): + def __init__(self, *, max_records: int, max_bytes: int) -> None: + super().__init__(max_records=max_records, max_bytes=max_bytes) + self.delivered = 0 + + def _send_batch(self, batch: Sequence[LogRecord]) -> None: + self.delivered += len(batch) + + +@_settings +@given(metas=st.lists(meta_dicts, max_size=30)) +def test_batching_transport_buffer_stays_bounded_by_count_and_bytes( + metas: list[dict[str, Any]], +) -> None: + transport = _Recording(max_records=4, max_bytes=1_500) + logger = Logger("app", transports=[transport]) + + for meta in metas: + logger.info("adversarial", **meta) + assert len(transport._buffer) < 4 + assert transport._buffer_bytes < 1_500 + + transport.close() + assert transport.delivered == len(metas) diff --git a/tests/test_transports/test_aiohttp_sender.py b/tests/test_transports/test_aiohttp_sender.py new file mode 100644 index 0000000..a534812 --- /dev/null +++ b/tests/test_transports/test_aiohttp_sender.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any + +import pytest + +pytest.importorskip("aiohttp") + +from logquill import HTTPTransport, Logger # noqa: E402 +from logquill.transports.aiohttp_sender import AiohttpSender # noqa: E402 + + +class _Collector: + def __init__(self) -> None: + self.bodies: list[bytes] = [] + self.content_types: list[str | None] = [] + self.client_ports: list[int] = [] + self.status = 200 + + +@pytest.fixture() +def server() -> Iterator[tuple[str, _Collector]]: + collector = _Collector() + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" # keep-alive + + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers["Content-Length"]) + collector.bodies.append(self.rfile.read(length)) + collector.content_types.append(self.headers["Content-Type"]) + collector.client_ports.append(self.client_address[1]) + self.send_response(collector.status) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, format: str, *args: Any) -> None: # silence + pass + + httpd = HTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread( + target=httpd.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ) + thread.start() + try: + yield f"http://127.0.0.1:{httpd.server_port}/ingest", collector + finally: + httpd.shutdown() + httpd.server_close() + + +def test_posts_a_batch_as_ndjson(server: tuple[str, _Collector]) -> None: + url, collector = server + sender = AiohttpSender(timeout=5.0) + try: + sender(url, ['{"a":1}', '{"a":2}']) + finally: + sender.close() + + assert collector.bodies == [b'{"a":1}\n{"a":2}'] + assert collector.content_types == ["application/x-ndjson"] + + +def test_reuses_one_connection_across_batches(server: tuple[str, _Collector]) -> None: + url, collector = server + sender = AiohttpSender(timeout=5.0) + try: + for _ in range(3): + sender(url, ["{}"]) + finally: + sender.close() + + assert len(collector.bodies) == 3 + assert len(set(collector.client_ports)) == 1 # one TCP connection, not three + + +def test_an_error_status_raises(server: tuple[str, _Collector]) -> None: + url, collector = server + collector.status = 500 + sender = AiohttpSender(timeout=5.0) + try: + with pytest.raises(Exception, match="500"): + sender(url, ["{}"]) + finally: + sender.close() + + +def test_an_unreachable_endpoint_raises_and_the_sender_recovers( + server: tuple[str, _Collector], +) -> None: + url, collector = server + sender = AiohttpSender(timeout=2.0) + try: + with pytest.raises(Exception): # noqa: B017 - aiohttp's ClientConnectorError + sender("http://127.0.0.1:1/ingest", ["{}"]) + sender(url, ["{}"]) + finally: + sender.close() + + assert len(collector.bodies) == 1 + + +def test_close_is_idempotent_and_a_later_call_starts_a_fresh_session( + server: tuple[str, _Collector], +) -> None: + url, collector = server + sender = AiohttpSender(timeout=5.0) + sender(url, ["{}"]) + sender.close() + sender.close() + + sender(url, ["{}"]) + sender.close() + + assert len(collector.bodies) == 2 + + +def test_http_transport_with_the_aiohttp_backend_delivers_records( + server: tuple[str, _Collector], +) -> None: + url, collector = server + logger = Logger("app", transports=[HTTPTransport(url, backend="aiohttp", batch_size=2)]) + + logger.info("one") + logger.info("two") + logger.close() + + lines = b"\n".join(collector.bodies).splitlines() + assert len(lines) == 2 + assert b'"message":"one"' in lines[0] diff --git a/tests/test_transports/test_http_transport.py b/tests/test_transports/test_http_transport.py index 1fedf85..788982a 100644 --- a/tests/test_transports/test_http_transport.py +++ b/tests/test_transports/test_http_transport.py @@ -1,5 +1,9 @@ +import logging +import sys from typing import List, Sequence, Tuple +import pytest + from logquill.logger import Logger from logquill.transports.http_transport import HTTPTransport @@ -46,3 +50,63 @@ def test_close_on_empty_batch_sends_nothing() -> None: transport.close() assert sender.calls == [] + + +def test_flushes_early_when_the_buffered_bytes_reach_max_bytes() -> None: + sender = FakeSender() + transport = HTTPTransport( + "https://example.com/logs", batch_size=1000, max_bytes=100, sender=sender + ) + logger = Logger("app.test", transports=[transport]) + + logger.info("big", blob="x" * 200) + + assert len(sender.calls) == 1 + + +def test_a_failing_sender_is_logged_not_raised_and_later_records_still_flow( + caplog: pytest.LogCaptureFixture, +) -> None: + calls: List[int] = [] + + def flaky(url: str, batch: Sequence[str]) -> None: + calls.append(len(batch)) + if len(calls) == 1: + raise OSError("connection refused") + + transport = HTTPTransport("https://example.com/logs", batch_size=1, sender=flaky) + + with caplog.at_level(logging.ERROR, logger="logquill"): + transport.write("first", None) # type: ignore[arg-type] + transport.write("second", None) # type: ignore[arg-type] + transport.close() + + assert calls == [1, 1] + assert "couldn't deliver 1 log record(s) to https://example.com/logs" in caplog.text + + +def test_close_releases_a_sender_that_holds_a_connection() -> None: + class ClosableSender(FakeSender): + closed = False + + def close(self) -> None: + self.closed = True + + sender = ClosableSender() + HTTPTransport("https://example.com/logs", sender=sender).close() + + assert sender.closed is True + + +def test_rejects_an_unknown_backend() -> None: + with pytest.raises(ValueError, match="backend must be"): + HTTPTransport("https://example.com/logs", backend="curl") # type: ignore[arg-type] + + +def test_aiohttp_backend_without_aiohttp_gives_an_install_hint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setitem(sys.modules, "aiohttp", None) + + with pytest.raises(ImportError, match=r"pip install logquill\[http\]"): + HTTPTransport("https://example.com/logs", backend="aiohttp") diff --git a/tests/test_worker.py b/tests/test_worker.py index bd0d01e..50726fa 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import threading import time @@ -203,3 +204,22 @@ async def test_drain_async_awaits_completion_without_blocking_the_event_loop() - assert await worker.drain_async(timeout=2.0) is True assert sorted(seen) == list(range(20)) worker.close() + + +def test_the_first_drop_warns_even_when_the_monotonic_clock_is_still_near_zero( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + # the monotonic clock's zero point is arbitrary (boot time on many + # platforms), so a process started shortly after boot must still warn + monkeypatch.setattr("logquill.worker.time.monotonic", lambda: 5.0) + gate = threading.Event() + worker = AsyncWorker(max_queue_size=1, backpressure="drop_newest") + + with caplog.at_level(logging.WARNING, logger="logquill"): + worker.submit(gate.wait) # occupies the worker thread + worker.submit(lambda: None) # fills the queue + worker.submit(lambda: None) # dropped + gate.set() + worker.close() + + assert any("queue full" in record.getMessage() for record in caplog.records)