Packages
snakepit
0.9.0
0.13.0
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.1
0.9.0
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.11
0.6.10
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.1
0.2.0
0.1.2
0.1.1
0.1.0
High-performance pooler and session manager for external language integrations. Supports Python, Node.js, Ruby, and more with gRPC streaming, session management, and production-ready process cleanup.
Current section
Files
Jump to
Current section
Files
priv/python/snakepit_bridge/telemetry/backends/stderr.py
"""Stderr telemetry backend.
This backend writes telemetry events to stderr as JSON, which can be captured
by Elixir. This is a fallback mechanism for adapters that don't support gRPC.
"""
from __future__ import annotations
import json
import sys
import time
from typing import Any, Dict, Optional
from .base import TelemetryBackend
class StderrBackend(TelemetryBackend):
"""Telemetry backend that writes events to stderr as JSON.
Events are prefixed with "TELEMETRY:" for easy parsing by Elixir.
Example output:
TELEMETRY:{"event":"tool.execution.start","measurements":{"system_time":1698234567890},"metadata":{"tool":"predict"},"timestamp_ns":1698234567890123456,"correlation_id":"abc-123"}
"""
def emit(
self,
event_name: str,
measurements: Dict[str, Any],
metadata: Optional[Dict[str, Any]] = None,
correlation_id: Optional[str] = None,
) -> None:
"""Emit a telemetry event to stderr.
Args:
event_name: Event name in dotted notation
measurements: Numeric measurements
metadata: Contextual metadata
correlation_id: Optional correlation ID
"""
payload = {
"event": event_name,
"measurements": measurements,
"metadata": metadata or {},
"timestamp_ns": time.time_ns(),
"correlation_id": correlation_id or "",
}
try:
sys.stderr.write(f"TELEMETRY:{json.dumps(payload)}\n")
sys.stderr.flush()
except Exception as e:
# Don't crash the worker if telemetry fails
sys.stderr.write(f"TELEMETRY_ERROR: {e}\n")
sys.stderr.flush()