orchestrata_ultimate.core.monitoring.spans

Reference › orchestrata_ultimate › core › monitoring › spans

The data model of execution traces. A Span is a slotted dataclass for one function invocation: qualified func_name, an optional grouping label, monotonic start_ns and end_ns, a status, an error type and message (the message is the first 500 characters of the exception text), the OS thread_id, and the source file and line of the definition. The properties duration_ns, duration_us and duration_ms give the duration, and to_dict serialises a span for logging or graph storage. SpanStatus has three values, OK, ERROR and TIMEOUT. _make_span creates a span with the start time set to now. The engine creates and stores the spans.

At a glance

Class diagram

Dependency diagram

API

Classes

SpanStatus (enum)

Bases: enum.Enum · Line 12

Outcome of a traced function invocation.

Span (dataclass)

Decorators: dataclass(slots=True) · Line 21

A single traced function invocation.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/monitoring/spans.py

"""Span data model for execution traces.

A Span represents a single function invocation — entry, exit, duration,
and optional error information. Spans are the atoms of the tracing system.
"""

import enum
import time
from dataclasses import dataclass


class SpanStatus(enum.Enum):
    """Outcome of a traced function invocation."""

    OK = "ok"
    ERROR = "error"
    TIMEOUT = "timeout"


@dataclass(slots=True)
class Span:
    """A single traced function invocation.

    Attributes:
        func_name: Qualified function name (e.g., 'module.Class.method').
        label: Optional human-readable label for grouping.
        start_ns: Monotonic clock timestamp at entry (nanoseconds).
        end_ns: Monotonic clock timestamp at exit (nanoseconds).
        status: Whether the function returned normally or raised.
        error_type: Exception class name if status is ERROR.
        error_message: First 500 chars of exception str if ERROR.
        thread_id: OS thread ID that executed the function.
        code_file: Source file path.
        code_line: Line number of the function definition.
    """

    func_name: str
    label: str = ""
    start_ns: int = 0
    end_ns: int = 0
    status: SpanStatus = SpanStatus.OK
    error_type: str = ""
    error_message: str = ""
    thread_id: int = 0
    code_file: str = ""
    code_line: int = 0

    @property
    def duration_ns(self) -> int:
        """Wall-clock duration in nanoseconds."""
        return self.end_ns - self.start_ns

    @property
    def duration_ms(self) -> float:
        """Wall-clock duration in milliseconds."""
        return self.duration_ns / 1_000_000

    @property
    def duration_us(self) -> float:
        """Wall-clock duration in microseconds."""
        return self.duration_ns / 1_000

    def to_dict(self) -> dict:
        """Serialize to dictionary for logging or graph storage."""
        return {
            "func_name": self.func_name,
            "label": self.label,
            "duration_ms": round(self.duration_ms, 3),
            "status": self.status.value,
            "error_type": self.error_type,
            "error_message": self.error_message,
            "thread_id": self.thread_id,
            "code_file": self.code_file,
            "code_line": self.code_line,
        }


def _make_span(
    func_name: str,
    label: str,
    code_file: str,
    code_line: int,
    thread_id: int,
) -> Span:
    """Create a span with start_ns set to now."""
    return Span(
        func_name=func_name,
        label=label,
        start_ns=time.monotonic_ns(),
        thread_id=thread_id,
        code_file=code_file,
        code_line=code_line,
    )