mgf-logship

Centralized log shipping for the Magogi federation — the Layer-B half of the unified-logging program (see mgf-common/docs/inprogress/UNIFIED_LOGGING_DESIGN.md, decision D4: ship + protocol now, stub the server).

A consumer attaches LogShipper to its logging tree; it batches the canonical mgf-common LogRecord JSON (the same shape every runtime emits — OB-10), redacts each record, and POSTs them to a collector over a versioned, language-neutral wire protocol. A store-now stub collector (FastAPI + sqlite) is included so the protocol works end-to-end today; analysis and presentation are a later program.

Install

pip install mgf-logship                # shipper only (stdlib transport)
pip install mgf-logship[collector]     # + the FastAPI/sqlite stub collector

Ship logs from an app

import logging
from mgf.logship import LogShipper
from mgf.common.observability import current_log_stats  # optional stats shipping

shipper = LogShipper(
    "https://collector.internal",
    token="…",                       # Authorization: Bearer …
    stats_provider=current_log_stats,  # also ship LogStats every 30s
)
logging.getLogger().addHandler(shipper)
# … on shutdown: shipper.close()  (flushes the queue)

LogShipper is fail-open (a slow/down collector never blocks or crashes the app), bounded (the queue is capped; oldest dropped on overflow, counted in dropped_count), and redacts before ship (off-box data is scrubbed).

In production (env-driven, with app context)

The pattern the fleet uses (reference: mgf.fed._logging.bootstrap_logging, the mgf-fed CLI):

import logging, os
from mgf.common import bootstrap   # canonical app bootstrap (mgf-common)
from mgf.logship import LogShipper

def bootstrap_logging() -> None:
    url, token = os.environ.get("MGF_LOGSHIP_URL"), os.environ.get("MGF_LOGSHIP_INGEST_TOKEN")
    if not (url and token):
        return                      # opt-in: unconfigured ⇒ no-op, app unchanged
    bootstrap(app_name="myapp", app_version=__version__)   # ← see note below
    logging.getLogger().addHandler(
        LogShipper(url, token=token, app="myapp", env=os.environ.get("MGF_LOGSHIP_ENV", "prod"))
    )

• Call mgf.common.bootstrap() first. The shipper enriches each record via mgf.common.app_version() / current_context(); without an active app context mgf-common warns (current_context() called before bootstrap()) and records ship without app/trace context. bootstrap() also installs the standard console+file sinks.
• MGF_LOGSHIP_URL is the collector base (e.g. https://<host>.sslip.io) — the shipper appends /ingest itself.
• Flush on exit: the bounded queue flushes via LogShipper.close(), which Python's logging.shutdown (registered at interpreter exit) calls — so even a short-lived CLI run drains before exit. No extra wiring needed.
• Keep it guarded. Make the LogShipper/mgf.common imports lazy and failure-silent if your package has a dependency-light core; logging setup must never crash the app.

Run the stub collector

from mgf.logship.collector import create_collector_app
app = create_collector_app(db_path="logship.db", token="…")
# uvicorn module:app

Endpoints: POST /ingest (NDJSON of LogRecords), POST /stats (a LogStats JSON), GET /health. It persists to sqlite and stops there.

The wire protocol

mgf.logship._protocol is the source of truth: SHIP_PROTOCOL_VERSION, the endpoint paths, the X-Logship-* headers, and the NDJSON framing. Deliberately HTTP + NDJSON so a Zig or TS shipper can talk to the same collector with nothing but a socket and a JSON encoder.

License

MIT — see LICENSE.