stemtrace 0.3.3

A lightweight Celery task flow visualizer

stemtrace 🌿

Zero-infrastructure Celery task flow visualizer

---

Flower shows you what exists. Stemtrace shows you what happened.

Ever stared at a failed Celery task wondering "what called this?" or "why did it retry 5 times?"

Stemtrace captures your task executions as a graph — visualize parent→child flows, see retry chains, track groups and chords, all without adding any new infrastructure.

stemtrace supports Redis and RabbitMQ for event transport.

✨ Features

See What Happened

• Task Flow Graphs — Visualize chains, groups, and chords as interactive DAGs
• Execution Timeline — Track queued → started → retried → finished states
• Arguments & Results — Inspect inputs, outputs, and exceptions
• Retry Chains — Understand exactly when and why retries happened

Canvas Support

• Groups & Chords — Automatic visualization of group() and chord() patterns
• Parent-Child Tracking — See which task spawned which

Worker Monitoring & Registry

• Workers page — See which workers are online/offline and what tasks they have registered
• Registry status badges — Quickly spot tasks that are active, never run, or not registered by any current worker

Production Ready

• Zero Infrastructure — Uses your existing broker (Redis or RabbitMQ), no database needed
• Sensitive Data Scrubbing — Passwords and API keys filtered automatically
• Read-Only — Safe for production; never modifies your task queue
• FastAPI Integration — Mount into your existing app with one line

🔎 What you’ll see in the dashboard

Task details (timing, inputs/outputs, errors)

• What you’ll see: Per-task execution timing (including how long it spent in each state), parameters (args/kwargs), return value, and the full event history.
• Why it helps: Quickly answer “what happened?” for a single task: slow queueing vs slow execution, which retry succeeded, and (on failures) the exception + traceback for debugging.

Flow graphs (chains, groups, chords)

• What you’ll see: An interactive DAG of your workflow with parent→child edges, plus clear GROUP/CHORD containers for Celery canvas patterns.
• Why it helps: Understand fan-out/fan-in at a glance (especially chords), spot which branch failed, and debug “why didn’t my callback run?” without grepping logs.

Task registry (registration status + warnings)

• What you’ll see: A registry of tasks with status badges like Active, Never Run, and Not Registered plus “registered by …” worker info.
• Why it helps: Catch misconfigurations where tasks get stuck in PENDING because no current worker has the task registered (common in multi-repo or deploy drift scenarios).

🚀 Quick Start

1. Install

# Using pip
pip install stemtrace

# Using uv
uv add stemtrace

2. Instrument your Celery app

from celery import Celery
import stemtrace

app = Celery("myapp", broker="redis://localhost:6379/0")

# One line to enable event capture.
# Tip: put this in the module where you define your Celery app so it's imported by
# both Celery workers and any code that calls app.send_task()/delay().
stemtrace.init_worker(app)

3. View the dashboard

Option A: Standalone server (new container/process)

stemtrace server

Open http://localhost:8000.

Tip: make sure the server is pointed at the same broker as your workers:

stemtrace server --broker-url redis://localhost:6379/0
# or:
stemtrace server --broker-url amqp://guest:guest@localhost:5672//

Option B: Embed in your FastAPI app (no extra container)

from fastapi import FastAPI
import stemtrace

app = FastAPI(lifespan=my_lifespan)  # Your existing app

stemtrace.init_app(app, broker_url="redis://localhost:6379/0")

Access at /stemtrace/ in your existing app — no new services to deploy.

See Deployment Options for auth, scaling, and more.

📦 Architecture

stemtrace is designed as two decoupled components:

┌──────────────────────────────────────────────────────────────────┐
│                        Your Application                          │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐        │
│  │ Celery Worker│    │ Celery Worker│    │ Celery Worker│        │
│  │ + stemtrace  │    │ + stemtrace  │    │ + stemtrace  │        │
│  └──────┬───────┘    └──────┬───────┘    └──────┬───────┘        │
│         │                   │                   │                │
│         └───────────────────┼───────────────────┘                │
│                             │ events                             │
│                             ▼                                    │
│                     ┌───────────────┐                            │
│                     │    Broker     │                            │
│                     └───────┬───────┘                            │
│                             │                                    │
└─────────────────────────────┼────────────────────────────────────┘
                              │
                              ▼
                    ┌───────────────────┐
                    │    stemtrace      │
                    │  server (viewer)  │
                    │  ┌─────────────┐  │
                    │  │   Web UI    │  │
                    │  └─────────────┘  │
                    └───────────────────┘

Library (stemtrace)

• Hooks into Celery signals
• Captures task lifecycle events
• Sends normalized events to the broker
• Zero overhead in critical path — fire-and-forget writes

Server (stemtrace server)

• Reads events from the broker
• Builds task graphs
• Serves the web UI
• Completely read-only — safe for production

🔧 Configuration

Library Options

import stemtrace

stemtrace.init_worker(
    app,
    # Optional: override broker URL (defaults to Celery's broker_url)
    transport_url="redis://localhost:6379/0",
    prefix="stemtrace",                        # Key/queue prefix
    ttl=86400,                                 # Event TTL in seconds (default: 24h)

    # Data capture (all enabled by default)
    capture_args=True,                         # Capture task args/kwargs
    capture_result=True,                       # Capture return values

    # Sensitive data scrubbing (Sentry-style)
    scrub_sensitive_data=True,                 # Scrub passwords, API keys, etc.
    additional_sensitive_keys=frozenset({"my_secret"}),  # Add custom keys
    safe_keys=frozenset({"public_key"}),       # Never scrub these keys

    # UI display
    node_alias_from_arguments="operator_type", # Use kwargs["operator_type"] as node name
    # node_alias_from_arguments="0",           # Or use args[0] (digit string = positional index)
)

# Introspection (after init)
stemtrace.is_initialized()   # -> True
stemtrace.get_config()       # -> StemtraceConfig
stemtrace.get_transport()    # -> EventTransport (for testing)

Sensitive Data Scrubbing

By default, stemtrace scrubs common sensitive keys from task arguments:

• Passwords: password, passwd, pwd, secret
• API keys: api_key, apikey, token, bearer, authorization
• Financial: credit_card, cvv, ssn
• Session: cookie, session, csrf

Scrubbed values appear as [Filtered] in the UI.

Canvas Graph Visualization

stemtrace automatically detects and visualizes Celery canvas constructs:

# Parent-spawned group: GROUP is child of parent
batch_processor
└── ┌─ GROUP ──────────┐
    │  ├── add(1, 2)   │
    │  ├── add(3, 4)   │
    │  └── add(5, 6)   │
    └──────────────────┘

# Standalone group: GROUP is a root node
┌─ GROUP ──────────┐
│  ├── add(1, 1)   │
│  ├── add(2, 2)   │
│  └── add(3, 3)   │
└──────────────────┘

# Chord: header tasks inside, callback outside with edges
┌─ CHORD ──────────┐
│  ├── add(10, 10) │──┐
│  ├── add(20, 20) │──┼──► aggregate_results
│  └── add(30, 30) │──┘
└──────────────────┘

• Synthetic containers — GROUP/CHORD nodes are always created when 2+ tasks share a group_id
• Parent linking — When spawned from a parent task, the container becomes a child of that parent
• Chord callbacks — Rendered outside the container with edges from each header task
• Timing — Each node displays start time and duration directly in the graph
• Aggregate state — Container shows running/success/failure based on member states

Environment Variables (Optional)

You do not need these environment variables if you’re using the Python APIs:

• stemtrace.init_worker(app, transport_url=..., prefix=..., ttl=...)
• stemtrace.init_app(app, broker_url=..., transport_url=...)

They exist mainly as convenient defaults for:

• stemtrace server / stemtrace consume
• container/Docker setups where passing flags is awkward

If your app already uses env vars like BROKER_URL, REDIS_URL, etc., just pass them through:

import os
import stemtrace

stemtrace.init_app(
    app,
    broker_url=os.environ["BROKER_URL"],
    transport_url=os.getenv("STEMTRACE_TRANSPORT_URL") or os.environ["BROKER_URL"],
)

Variable	Description	Default
STEMTRACE_BROKER_URL	Celery broker URL (used for on-demand worker/registry inspection). Also used as the default for STEMTRACE_TRANSPORT_URL.	redis://localhost:6379/0
STEMTRACE_TRANSPORT_URL	Event transport URL (where stemtrace publishes/consumes events).	Defaults to STEMTRACE_BROKER_URL.

Supported Brokers

Broker	URL Scheme	Status
Redis	redis://, rediss://	✅ Supported
RabbitMQ	amqp://, amqps://, pyamqp://	✅ Supported

Event Retention & Server Restarts (Important)

stemtrace builds the UI from events and keeps state in an in-memory graph store. That means a stemtrace server restart starts from an empty store and only becomes “full” again once events are re-consumed.

• Redis (Streams): On restart, the server can rebuild state by replaying events that are still retained in the stream (bounded by ttl / stream trimming).
• RabbitMQ (fanout + per-consumer queue):
  • Events already consumed/acked by the server are gone.
  • Events published while the server is down are only visible after restart if the server’s durable per-consumer queue still exists and the messages are still within TTL.
• Workers + Registry tabs: stemtrace uses Celery inspect on demand to populate workers and registered tasks, so those pages work even if the server missed worker_ready events.
• If you need durable history across restarts: point stemtrace events at Redis even if your Celery broker is RabbitMQ (set transport_url in stemtrace.init_worker(...) and set STEMTRACE_TRANSPORT_URL for the server / embedded setup).

🐳 Docker

docker run -p 8000:8000 \
    -e STEMTRACE_BROKER_URL=redis://host.docker.internal:6379/0 \
    ghcr.io/iansokolskyi/stemtrace

RabbitMQ example:

docker run -p 8000:8000 \
    -e STEMTRACE_BROKER_URL=amqp://guest:[e-mail]:5672// \
    ghcr.io/iansokolskyi/stemtrace

RabbitMQ broker + Redis event transport (recommended if you want history across server restarts):

docker run -p 8000:8000 \
    -e STEMTRACE_BROKER_URL=amqp://guest:[e-mail]:5672// \
    -e STEMTRACE_TRANSPORT_URL=redis://host.docker.internal:6379/0 \
    ghcr.io/iansokolskyi/stemtrace

Or with Docker Compose:

services:
  stemtrace:
    image: ghcr.io/iansokolskyi/stemtrace
    ports:
      - "8000:8000"
    environment:
      - STEMTRACE_BROKER_URL=redis://redis:6379/0

For a local RabbitMQ setup, see docker-compose.rabbitmq.yml.

🖥️ Deployment Options

stemtrace offers two deployment modes depending on your needs:

Mode	Best For	Command
Standalone Server	Dedicated monitoring, simple setup	stemtrace server
FastAPI Embedded	Single-app deployment, existing FastAPI apps	stemtrace.init_app(...)

Option 1: Standalone Server (Recommended)

The simplest way to run stemtrace — a dedicated monitoring service:

pip install stemtrace

stemtrace server

Open http://localhost:8000 to view the dashboard.

Server Options

stemtrace server \
    --broker-url redis://myredis:6379/0 \
    --transport-url redis://myredis:6379/0 \
    --host 0.0.0.0 \
    --port 8000 \
    --reload  # For development

Protecting the Server (Built-in Login Page)

stemtrace server \
    --broker-url redis://myredis:6379/0 \
    --login-username admin \
    --login-password secret \
    --login-secret change-me

High-Scale Production Setup

Note: stemtrace server includes an embedded consumer today (single-process). A multi-process deployment mode is planned.

Option 2: FastAPI Embedded

Mount stemtrace directly into your existing FastAPI application:

from fastapi import FastAPI
import stemtrace

app = FastAPI(lifespan=my_lifespan)  # Your existing app with lifespan

stemtrace.init_app(app, broker_url="redis://localhost:6379/0")  # Wraps lifespan, adds /stemtrace routes

That's it. init_app() automatically:

• Wraps your existing lifespan (Sentry, DB connections, etc. keep working)
• Mounts the dashboard at /stemtrace/
• Starts the event consumer

Tip: you can also set transport_url if you want stemtrace events stored separately from your Celery broker.

Configuration Options

import stemtrace

# Returns the underlying StemtraceExtension if you need it (optional).
extension = stemtrace.init_app(
    app,
    broker_url="redis://localhost:6379/0",
    transport_url=None,          # Defaults to broker_url
    prefix="/stemtrace",        # Mount path AND event stream prefix (normalized)
    ttl=86400,                  # Event TTL in seconds
    max_nodes=10000,            # Max nodes in memory
    embedded_consumer=True,     # Run consumer in FastAPI process
    serve_ui=True,              # Serve React dashboard
    auth_dependency=None,       # Optional auth (see below)
)

With Custom Authentication

from fastapi import Depends
import stemtrace
from your_app.auth import require_admin

stemtrace.init_app(app, broker_url="redis://localhost:6379/0", auth_dependency=Depends(require_admin))

With Built-in Login Page (Recommended for UI-first)

If you primarily use the UI, the easiest way to protect UI + assets + API + WebSocket is the built-in form login (cookie session):

import stemtrace

stemtrace.init_app(
    app,
    broker_url="redis://localhost:6379/0",
    login_username="admin",
    login_password="secret",
    login_secret="change-me",  # recommended for production
)

This serves a sign-in page at /stemtrace/login and protects everything under /stemtrace.

Built-in Auth Helpers (Basic / API key)

import stemtrace

stemtrace.init_app(
    app,
    broker_url="redis://localhost:6379/0",
    auth_dependency=stemtrace.require_basic_auth("admin", "secret"),
)

Embedded Consumer Modes

Mode	Use Case	Setup
Embedded	Development, simple apps	Default — consumer runs in FastAPI process
External	Production, high scale	Planned

🗺️ Roadmap

What's Working Now

• ✅ Task flow graphs — Visualize chains, groups, and chords as DAGs
• ✅ Full lifecycle tracking — PENDING → RECEIVED → STARTED → SUCCESS/FAILURE
• ✅ Canvas awareness — Automatic GROUP/CHORD node visualization
• ✅ Arguments & results — View inputs, outputs, and exceptions
• ✅ Sensitive data scrubbing — Passwords and API keys filtered automatically
• ✅ Real-time updates — WebSocket-powered live dashboard
• ✅ FastAPI integration — Mount into your existing app
• ✅ RabbitMQ support — Use your existing RabbitMQ broker (amqp://, amqps://, pyamqp://)
• ✅ Workers page — Monitor online/offline workers and their registered tasks
• ✅ Task registry — Browse discovered + registered tasks with clear status badges

Coming Soon

• 🔜 Anomaly detection — Spot stuck, orphaned, or failed tasks
• 🔜 Dashboard with stats — Success rates, durations, failure trends
• 🔜 OpenTelemetry export — Send traces to Jaeger, Tempo, Datadog
• 🔜 Webhook notifications — Push events to your systems
• 🔜 Data export — Download execution history as JSON

🤝 Contributing

Contributions, bug reports, and feature requests are welcome! This is a community project — if stemtrace helps you debug Celery, consider helping make it better.

See our Contributing Guide to get started.

git clone https://github.com/iansokolskyi/stemtrace.git
cd stemtrace
uv sync --extra dev   # Install dependencies
make check            # Run tests

📄 License

MIT — use it however you like.