Skip to main content

observability.storage

Trace storage for agent-runtimes observability.

Provides:

  • SQLite-based trace storage
  • Trace querying and filtering
  • Automatic cleanup of old traces

TraceSpan Objects​

class TraceSpan(BaseModel)

Represents a single trace span.

TraceQuery Objects​

class TraceQuery(BaseModel)

Query parameters for trace search.

TraceStorage Objects​

class TraceStorage(ABC)

Abstract base class for trace storage.

store_span​

@abstractmethod
def store_span(span: TraceSpan) -> None

Store a trace span.

get_span​

@abstractmethod
def get_span(trace_id: str, span_id: str) -> TraceSpan | None

Get a specific span by trace and span ID.

query_spans​

@abstractmethod
def query_spans(query: TraceQuery) -> list[TraceSpan]

Query spans based on criteria.

get_trace​

@abstractmethod
def get_trace(trace_id: str) -> list[TraceSpan]

Get all spans for a trace.

delete_trace​

@abstractmethod
def delete_trace(trace_id: str) -> bool

Delete a trace and all its spans.

cleanup_old_traces​

@abstractmethod
def cleanup_old_traces(retention_days: int) -> int

Delete traces older than retention period. Returns count deleted.

SQLiteTraceStorage Objects​

class SQLiteTraceStorage(TraceStorage)

SQLite-based trace storage implementation.

__init__​

def __init__(db_path: str | Path, max_traces: int = 10000)

Initialize SQLite trace storage.

Arguments:

  • db_path - Path to the SQLite database file.
  • max_traces - Maximum number of traces to retain.

store_span​

def store_span(span: TraceSpan) -> None

Store a trace span.

get_span​

def get_span(trace_id: str, span_id: str) -> TraceSpan | None

Get a specific span by trace and span ID.

query_spans​

def query_spans(query: TraceQuery) -> list[TraceSpan]

Query spans based on criteria.

get_trace​

def get_trace(trace_id: str) -> list[TraceSpan]

Get all spans for a trace.

delete_trace​

def delete_trace(trace_id: str) -> bool

Delete a trace and all its spans.

cleanup_old_traces​

def cleanup_old_traces(retention_days: int) -> int

Delete traces older than retention period.

get_stats​

def get_stats() -> dict[str, Any]

Get storage statistics.

SQLiteSpanExporter Objects​

class SQLiteSpanExporter()

OpenTelemetry SpanExporter that writes to SQLite storage.

__init__​

def __init__(db_path: str | Path, max_traces: int = 10000)

Initialize the SQLite span exporter.

Arguments:

  • db_path - Path to the SQLite database file.
  • max_traces - Maximum number of traces to retain.

export​

def export(spans: Sequence[Any]) -> Any

Export spans to SQLite storage.

shutdown​

def shutdown() -> None

Shutdown the exporter.

force_flush​

def force_flush(timeout_millis: int = 30000) -> bool

Force flush any pending spans.