Shadowbroker Logging and Monitoring Implementation for FastAPI Backend Services

Shadowbroker implements observability through standard Python logging configured in backend/main.py and a custom in-process metrics collector exposed via the /api/mesh/metrics endpoint.

Shadowbroker is a FastAPI-based Python backend service that relies on two complementary observability mechanisms to maintain operational visibility. The architecture combines Python’s built-in logging and monitoring capabilities with a lightweight custom metrics subsystem to track both diagnostic events and quantitative performance data without external dependencies.

Standard Python Logging Architecture

Root Logger Configuration in backend/main.py

The logging subsystem initializes at application startup through a concise configuration that sets the foundation for all subsequent log output. In backend/main.py, the service establishes a root logger with INFO level and creates module-specific child loggers:

# backend/main.py

logging.basicConfig(level=logging.INFO)            # ← sets the root logger level

logger = logging.getLogger(__name__)               # ← child logger for this module

Hierarchical Module Loggers

Each service module follows a consistent pattern of creating named loggers using __name__, producing hierarchical identifiers like backend.services.wormhole_supervisor and backend.services.tor_hidden_service. This design allows operators to filter and configure logging granularity per component. For example, in backend/services/wormhole_supervisor.py:

# backend/services/wormhole_supervisor.py

import logging
logger = logging.getLogger(__name__)   # → logger named "backend.services.wormhole_supervisor"

The logger captures operational events throughout the codebase:

logger.info("Loaded secret %s from %s", _var, _file_path)
logger.warning("Infonet private transport warmup incomplete: %s", tor_result)
logger.error("Failed to read secret file %s for %s: %s", _file_path, _var, _e)

Container-Ready Log Output

Because all log records write to stdout, the service remains stateless and compatible with container orchestration platforms. Docker, Kubernetes, and external aggregation stacks like ELK or Loki automatically capture these streams without requiring file handlers or persistent volumes.

Custom Mesh Metrics Collector

The mesh_metrics API

The services.mesh.mesh_metrics module provides a lightweight in-process collector offering three public functions for quantitative observability:

• increment(name: str) – Increments a named counter for event tracking
• observe_ms(name: str, value: float) – Records latency measurements in milliseconds
• snapshot() – Returns a JSON-serializable dictionary of all counters and timers

Modules import these helpers via aliases to minimize verbosity, as seen in backend/services/mesh/mesh_wormhole_dead_drop.py:

# backend/services/mesh/mesh_wormhole_dead_drop.py

from services.mesh.mesh_metrics import increment as metrics_inc
...
metrics_inc("alias_rotations_completed")

Security and Operational Counters

The collector tracks critical mesh events throughout the codebase. In backend/services/mesh/mesh_wormhole_dead_drop.py, counters monitor alias binding rejections:

metrics_inc("alias_bindings_rejected_revoked")
metrics_inc("alias_bindings_rejected_replay")

Meanwhile, backend/services/mesh/mesh_signed_events.py increments security-related metrics for envelope policy transitions and session restore failures. Timing metrics like ban_rotation_latency_ms use observe_ms to capture performance characteristics.

Metrics Endpoint and Access Control

The FastAPI router exposes metrics through GET /api/mesh/metrics, defined in the mesh router module (routers/mesh.py). This endpoint requires the gate.audit scope for authorization, as verified in the test suite, and returns structured data compatible with Prometheus scraping:

{
    "counters": {
        "session_restore_failures": 2,
        "envelope_policy_transitions": 1
    },
    "timers": {
        "ban_rotation_latency_ms": 42.5
    }
}

Practical Code Examples

Complete implementations demonstrate the integration of both systems. A service loading secrets combines error handling with hierarchical logging:

import logging
logger = logging.getLogger(__name__)

def load_secret(var_name: str, path: str) -> None:
    try:
        with open(path) as f:
            value = f.read().strip()
        logger.info("Loaded secret %s from %s", var_name, path)
    except FileNotFoundError:
        logger.error("Secret file %s for %s not found", path, var_name)

Metric instrumentation remains equally concise:

from services.mesh.mesh_metrics import increment as metrics_inc

def handle_alias_rotation():
    # … rotation logic …

    metrics_inc("alias_rotations_completed")

Testing utilities can verify metric collection directly:

from services.mesh.mesh_metrics import snapshot, increment

def test_metric_collection():
    increment("test_counter")
    snap = snapshot()
    assert snap["counters"]["test_counter"] == 1

Summary

Shadowbroker’s logging and monitoring implementation balances simplicity with observability requirements:

• Standard Python logging configured in backend/main.py provides hierarchical, module-specific loggers that write to stdout for container compatibility
• Per-module loggers in services like wormhole_supervisor.py and tor_hidden_service.py enable granular filtering and consistent formatting
• Custom mesh metrics in services.mesh.mesh_metrics offer lightweight counters and timers without external dependencies
• Secure exposure via /api/mesh/metrics with gate.audit scope authorization allows Prometheus scraping and manual health checks
• Stateless design ensures logs flow directly to Docker/Kubernetes collectors without file system dependencies

Frequently Asked Questions

Where is the logging level configured in Shadowbroker?

The root logger level is set to INFO in backend/main.py using logging.basicConfig(level=logging.INFO). This configuration applies to all child loggers created via logging.getLogger(__name__) throughout the service modules.

How does Shadowbroker expose metrics for external monitoring systems?

The backend exposes a JSON metrics snapshot through the FastAPI endpoint GET /api/mesh/metrics, implemented in the mesh router (routers/mesh.py). This endpoint returns counters and timers from the mesh_metrics collector and requires the gate.audit scope for access, making it compatible with Prometheus or custom health-check scripts.

What types of events does the mesh_metrics collector track?

The collector tracks both operational counters like alias_rotations_completed and envelope_policy_transitions, and security-related events such as alias_bindings_rejected_revoked and session_restore_failures. It also captures timing data for performance metrics like ban_rotation_latency_ms using the observe_ms function.

Why does Shadowbroker use stdout instead of file logging?

Writing logs to stdout maintains the service’s stateless architecture, allowing container runtimes like Docker and Kubernetes to capture and rotate logs automatically. This approach eliminates file handler complexity and enables seamless integration with centralized logging stacks like ELK or Grafana Loki without modifying application code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}