muga 0.1.4b0

Observability for developers who live in the terminal.

muga (Python SDK)

Muga observability SDK for Python. Ships logs to Muga via OpenTelemetry OTLP, captures uncaught exceptions, instruments Flask/FastAPI, and pings cron monitors.

Install

pip install muga

Framework extras:

pip install 'muga[flask]'
pip install 'muga[fastapi]'

Set your project token:

export MUGA_TOKEN=muga_xxxxxxxxxxxx

Quickstart

from muga import init, MugaLogger

init(service_name="api")

log = MugaLogger("api")
log.info("server started", {"port": "3000"})

init() reads MUGA_TOKEN and MUGA_ENDPOINT from env (or accepts token= / endpoint=). When the token is missing, init() does not raise — it prints a single warning to stderr and returns. MugaLogger calls and the framework middlewares stay importable; their emits become no-ops via OpenTelemetry's default NoOpLoggerProvider. This keeps dev/CI runs unblocked when no token is configured.

On successful init the resolved endpoint and service name are printed to stderr so misconfigurations (wrong MUGA_ENDPOINT, fallback to default) are visible without enabling debug=True.

Auto-exception capture

init() installs sys.excepthook, threading.excepthook, and an asyncio loop exception handler by default. They emit a FATAL/ERROR log with exception.type, exception.message, and exception.stacktrace attributes, force-flush the batch processor so the record drains before the interpreter exits, then delegate to the runtime's default behaviour.

Behaviour matrix

Source	Severity	Triggered by
sys.excepthook	FATAL	Uncaught exception on the main thread.
threading.excepthook	ERROR	Uncaught exception in a worker thread.
asyncio loop handler	ERROR	Unhandled exception on a task in a loop attached at init() time.

Caveats:

• The asyncio handler is only attached to the loop that exists when init() runs. If your code creates a new loop later (asyncio.run(...), a fresh asyncio.new_event_loop()), call install_exception_handlers(loop=...) from inside that loop, or re-call init() once the loop is current.
• All three hooks are idempotent — calling init() twice does not double-install.

Opt out:

init(capture_exceptions=False)

Flask

from flask import Flask
from muga import init
from muga.flask import muga_flask

init()
app = Flask(__name__)
muga_flask(app)

One log per request: INFO for <500, ERROR for >=500. Attributes: http.method, http.path, http.status_code, http.duration_ms, http.user_agent.

FastAPI

from fastapi import FastAPI
from muga import init
from muga.fastapi import MugaFastAPIMiddleware

init()
app = FastAPI()
app.add_middleware(MugaFastAPIMiddleware)

Same attribute set as the Flask middleware.

Cron heartbeat

Stand-alone helper. Does not require init().

from muga import heartbeat

heartbeat("daily-cleanup")

POSTs to <endpoint>/v1/crons/daily-cleanup/ping with Authorization: Bearer $MUGA_TOKEN. Raises on missing token or non-2xx response. The cron name is URL-encoded.

For async code, wrap in a thread:

import asyncio
from muga import heartbeat

await asyncio.to_thread(heartbeat, "daily-cleanup")

Shutdown

Flush pending logs on graceful shutdown:

import signal
from muga import shutdown

def _handle(*_):
    shutdown()
    raise SystemExit(0)

signal.signal(signal.SIGTERM, _handle)

Configuration reference

Argument	Env var	Default
token	MUGA_TOKEN	(missing → SDK no-ops with a stderr warning)
endpoint	MUGA_ENDPOINT	https://api.muga.sh
service_name	MUGA_SERVICE	default
capture_exceptions	—	True
debug	—	False
max_queue_size	—	10000
sampling_ratio	—	1.0
high_rate_warn_threshold	—	1000
high_rate_warn_window_s	—	60.0
max_retries	—	5

init() also registers atexit.register(shutdown) on first successful call so any process exit drains queued records. Repeat calls do not double-register.

Defenses against server pushback

A misbehaving paid client can flood the server with telemetry while OOM-ing itself. The SDK ships client-side defenses so a noisy host degrades gracefully instead of crashing or punching through quotas.

from muga import init

init(
    max_queue_size=10_000,         # bounded queue; drops oldest on overflow
    sampling_ratio=0.25,           # head-sample 25% of records before batching
    high_rate_warn_threshold=1000, # warn once per window above this rate
    high_rate_warn_window_s=60.0,
    max_retries=5,                 # retry budget for 429 / 5xx / transport errors
)

What's enabled by default:

• Retry-After honoured on 429. When the server returns 429, the exporter parses Retry-After (delta-seconds or HTTP-date) and waits that long before retrying.
• Exponential backoff with jitter on 5xx and transport errors. Per-attempt sleep is random() * min(cap, base * 2^attempt). After max_retries exhaustion the batch is dropped and a counter incremented.
• Bounded queue with drop_oldest policy. Each pipeline (logs, spans) gets its own bounded queue. On overflow the oldest record is dropped to free a slot for the newest, and dropped_count is incremented.
• High-rate warning, deduplicated. A sliding window counts records over high_rate_warn_window_s; once it crosses high_rate_warn_threshold, a single WARNING is logged via the standard logging module. The SDK does not log the warning again until the next window.
• Optional head sampling. sampling_ratio < 1.0 drops records before they enter the queue. For spans the decision is deterministic on trace_id so a given trace either keeps all of its spans or none.

Motivating scenario

A user ships v1 of a worker that loops over a 10-million-row table and emits one INFO log per row. Without these defenses the SDK would buffer all ten million records in memory (OOM-ing the worker), then flood the ingest endpoint while the server applies its own rate limiter and starts returning 429. With the defaults the SDK now (a) caps in-memory buffering at 10k, (b) drops the oldest records to keep the most recent context, (c) honours the server's Retry-After, and (d) prints one WARNING to the worker's log so the operator sees something is wrong.

Migration note

max_queue_size=10000 is a new default in this version. Previous releases used the OTel SDK default (unbounded). If your application explicitly relied on unbounded buffering — for example, a short-lived script that emits a burst at the very end and counts on every record reaching the server — pass an explicit large cap:

init(max_queue_size=10_000_000)

There is no separate "unbounded" mode; pick a number high enough to absorb your peak.

Reporting bugs

File a SDK bug report — the form prompts for SDK version, runtime, minimal repro, and expected vs actual. For anything potentially security-sensitive (token leaks, signature bypass, etc.), email [e-mail] instead of opening a public issue.