Skip to Content
ContributeTracing SystemArchitecture

Architecture

Detailed architecture of the Rhesis tracing system.

Component Architecture

SDK Components

OpenTelemetry Integration

The TracerProvider and BatchSpanProcessor are set up once, in the shared packages/rhesis/src/rhesis/telemetry/provider.py package (imported by both the SDK and the backend):

packages/rhesis/src/rhesis/telemetry/provider.py
resource = Resource.create({
    "service.name": service_name,
    "service.namespace": "rhesis",
    "deployment.environment": environment,
})

_TRACER_PROVIDER = TracerProvider(resource=resource)
exporter = RhesisOTLPExporter(api_key=api_key, base_url=base_url, project_id=project_id, environment=environment)

span_processor = BatchSpanProcessor(
    exporter,
    max_queue_size=2048,        # Max spans in memory
    max_export_batch_size=512,  # Spans per HTTP request
    schedule_delay_millis=5000, # Export every 5 seconds
)
_TRACER_PROVIDER.add_span_processor(span_processor)

Span Creation

@observe (sdk/src/rhesis/sdk/decorators/observe.py) wraps sync, async, and generator functions, starting a span on each call and setting attributes passed to the decorator. @endpoint (sdk/src/rhesis/sdk/decorators/endpoint.py) does the same and additionally registers the function for remote invocation over WebSocket.

Span names are validated server-side (packages/rhesis/src/rhesis/telemetry/schemas.py) against the pattern ai.<domain>.<action> (e.g. ai.llm.invoke) or function.<name>. The domains chain, workflow, and pipeline are rejected in favor of primitive operations; agent is allowed via the dedicated ai.agent.invoke / ai.agent.handoff operations for multi-agent tracing.

OTLP Exporter

RhesisOTLPExporter (packages/rhesis/src/rhesis/telemetry/exporter.py) subclasses OpenTelemetry’s OTLPSpanExporter:

Exporter Setup
class RhesisOTLPExporter(OTLPSpanExporter):
    def __init__(self, api_key, base_url, project_id, environment,
                 timeout=10, max_attempts=3, max_chunk_size=100):
        self.endpoint = f"{base_url.rstrip('/')}/telemetry/traces"
        super().__init__(endpoint=self.endpoint, timeout=timeout)
        self._session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })

export() converts OTEL spans to the shared OTELSpan schema, splits batches larger than max_chunk_size (100) into separate requests, and retries transient failures (connection errors, timeouts, 408/429/5xx) with exponential-jitter backoff bounded by a wall-clock deadline. Root spans are always placed in the first chunk. After 5 consecutive export failures it logs a warning with the endpoint and success rate.

Backend Components

Ingestion Endpoint

ingest_trace() in routers/telemetry.py (POST /telemetry/traces):

  1. Resolves project_id — prefers the span’s project_id, falls back to the request’s project scope, otherwise returns 422
  2. Validates the OTLP payload (Pydantic OTELTraceBatch, including span-name conventions)
  3. Stores spans via crud.create_trace_spans(), commits, and releases the DB session
  4. Dispatches post_ingest_link.delay(...) — fire-and-forget
Dispatch
try:
    post_ingest_link.delay(**dispatch_kwargs)
except BROKER_ERRORS as broker_err:
    logger.warning(f"Broker unavailable, post-processing deferred | error={broker_err}")

return TraceResponse(status="received", span_count=stored_count, trace_id=trace_id)

If the broker is unreachable, the request still returns 200 — linking and enrichment for that batch simply do not run. This differs from a separate code path used by direct SDK-endpoint invocation (services/invokers/tracing.py), which enqueues enrichment via AsyncService.execute_with_fallback: it tries the async Celery dispatch first and only falls back to running enrichment synchronously, in-process, when the broker raises a connection error.

The post_ingest_link task itself (tasks/telemetry/post_ingest.py) performs test-result linking, conversation-id linking, input-file linking, and dispatches an enrichment → evaluation chain per root span. See Worker: Trace Ingestion Pipeline for the full breakdown, including metric evaluation.

Linking Service

TraceLinkingService (services/telemetry/linking_service.py) has two entry points that share one implementation:

Linking Service
class TraceLinkingService:
    def link_traces_for_test_result(
        self, test_run_id, test_id, test_configuration_id, test_result_id, organization_id,
    ) -> int:
        """Called after test result creation (catches slow tests)."""
        return crud.update_traces_with_test_result_id(...)

    def link_traces_for_incoming_batch(self, spans, organization_id) -> int:
        """Called from post_ingest_link (catches fast tests)."""
        # Extracts test_run_id/test_id/test_configuration_id from the first
        # span's attributes, finds the matching TestResult, then links.

Enrichment

TraceEnricher (services/telemetry/enrichment/processor.py) computes costs, anomalies, and metadata from a trace’s spans:

  • Costs — calculate_token_costs() uses litellm.cost_per_token() on spans with ai.operation.type = llm.invoke, in USD and EUR
  • Anomalies — detect_anomalies() flags spans over 10 seconds (slow_span), LLM spans over 10,000 total tokens (high_token_usage), and error-status spans (error)
  • Metadata — extract_metadata() collects unique models, tools, and operation types, plus the root span’s name

The result is cached in the enriched_data JSONB column:

Enrichment Shape
{
  "costs": {
    "total_cost_usd": 0.023,
    "total_cost_eur": 0.021,
    "breakdown": [
      {
        "span_id": "...",
        "model_name": "gpt-4",
        "input_tokens": 150,
        "output_tokens": 80,
        "total_cost_usd": 0.023,
        "total_cost_eur": 0.021
      }
    ]
  },
  "anomalies": [
    {
      "type": "slow_span",
      "span_id": "...",
      "span_name": "ai.llm.invoke",
      "duration_ms": 12340,
      "message": "Span took 12.3s (threshold: 10s)"
    }
  ],
  "metrics": {
    "total_duration_ms": 12500,
    "span_count": 5,
    "error_count": 0
  },
  "models_used": [
    "gpt-4"
  ],
  "tools_used": [
    "search"
  ],
  "operation_types": [
    "llm.invoke",
    "tool.invoke"
  ],
  "root_operation": "ai.llm.invoke"
}

Re-enrichment is skipped once every span in the trace has a non-null processed_at — new child spans (later LLM calls, tool calls) trigger a re-run so multi-turn traces stay up to date as they arrive.

Query API

Trace Retrieval

Query Endpoint
@router.get("/traces/{trace_id}", response_model=TraceDetailResponse)
def get_trace(trace_id: str, project_id: str, db: Session, tenant_context):
    spans = crud.get_trace_by_id(db, trace_id, project_id, organization_id,
                                  eager_load=["project", "test_run", "test_result", "test"])
    root_spans = build_span_tree(spans)
    ...

GET /telemetry/traces/{trace_id} (project_id is a required query param) returns the span tree plus trace-level rollups and linked entities:

FieldDescription
root_spansSpans arranged as a parent-child tree
span_count, error_count, total_tokens, total_cost_usdComputed across all spans in the trace
trace_metrics_status, trace_reviewsLLM-based metric evaluation status and human review state
project, endpoint, test_run, test_result, testLinked entities, populated where applicable

Configuration

BatchSpanProcessor

SDK Configuration
BatchSpanProcessor(
    schedule_delay_millis=5000,    # Export every 5 seconds
    max_export_batch_size=512,     # Max spans per batch
    max_queue_size=2048,           # Queue size before forced export
)

Environment Variables

.env
# Backend database (component vars, not a single DATABASE_URL)
DB_HOST=localhost
DB_NAME=rhesis-db
APP_DB_USER=app-user
APP_DB_PASS=secure-password

# Celery broker
BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/1

# Enrichment (fallback rate; a live rate is fetched and cached normally)
USD_TO_EUR_RATE=0.92

# SDK
RHESIS_API_KEY=your_api_key
RHESIS_BASE_URL=http://localhost:8080
RHESIS_PROJECT_ID=your_project_id   # fallback when no project-scoped token is used

See Environment Variables for the full reference.

Celery Workers

Traces are processed by the same worker(s) that run test execution — see Worker: Background Tasks for queue and concurrency configuration.

Start Workers
celery -A rhesis.backend.worker.app worker --pool threads -n main@%h \
--queues=celery,execution,telemetry \
--concurrency=$CELERY_WORKER_CONCURRENCY \
--prefetch-multiplier=$CELERY_WORKER_PREFETCH_MULTIPLIER \
--optimization=fair
OverviewTrace Lifecycle