The SGP Tracing SDK offers flexible ways to instrument your application by creating traces and spans. You can choose between two main approaches: using context managers for automatic lifecycle management, or explicit control for manual start/end handling.

The most straightforward and robust way to create traces and spans is by using them as Python context managers (with statements). This approach automatically handles span start and end times, associates spans with the correct trace context, and captures exceptions, significantly reducing boilerplate and potential errors.

​
Creating a Trace with a Root Span

Every trace begins with a root span. Use tracing.create_trace() as a context manager to define a new top-level workflow. This automatically creates the root span for your trace, setting the context for all child spans created within its block.

import scale_gp_beta.lib.tracing as tracing

def my_workflow():
    # This creates a new trace and its root span
    with tracing.create_trace(name="my_application_workflow", metadata={"env": "production"}):
        print("Starting my application workflow...")
        # ... your main workflow logic goes here ...
        print("Application workflow completed.")

​
Creating Spans within a Trace

Inside a tracing.create_trace block, use tracing.create_span() as a context manager. These spans will automatically:

  • Inherit the trace_id from the current trace.
  • Be associated as a child of the currently active span (if another create_span is active) or the root span.
import time
import scale_gp_beta.lib.tracing as tracing

def fibonacci(curr: int) -> int:
    # This span becomes a child of the current active span (e.g., "main_execution")
    with tracing.create_span("fibonacci_calculation", input={"curr": curr}) as span:
        time.sleep(0.05) # Simulate some work
        if curr < 2:
            span.output = {"res": curr}
            return curr

        res = fibonacci(curr - 1) + fibonacci(curr - 2)
        span.output = {"res": res}
        return res

def main_traced_example():
    # This creates the main trace and its root span
    with tracing.create_trace("my_fibonacci_trace"):
        # This span will be a child of the "my_fibonacci_trace" root span
        with tracing.create_span("main_execution", metadata={"version": "1.0"}) as main_span:
            fib_result = fibonacci(5)
            main_span.output = {"final_fib_result": fib_result}
            print(f"Fibonacci(5) = {fib_result}")

​
Explicit Control

For advanced scenarios where context managers are not suitable (e.g., integrating with existing systems that manage context, or reporting historical data), you can manually manage the lifecycle of your spans. This approach requires more diligence to ensure start() and end() are always called, and consistency is maintained across your trace.

​
Manually Managing Spans (without implicit context)

You can create Span objects and explicitly control their trace_id and parent_id for fine-grained hierarchy management. This is particularly useful when you need to bridge tracing across different processes, services, or when a context manager is not practical.

Remember to manually call span.start() to begin the span’s lifecycle and span.end() to complete it. For error handling, you’ll need to wrap your logic in a try...finally block.

import uuid
import time
import random
from typing import Any, Dict
import scale_gp_beta.lib.tracing as tracing

class MockDatabase:
    def __init__(self) -> None:
        self._data = {
            "SELECT * FROM users WHERE id = 1;": {"id": 1, "name": "Alice"},
            "SELECT * FROM users WHERE id = 2;": {"id": 2, "name": "Bob"},
        }
    def execute_query(self, query: str, trace_id: str) -> Dict[str, Any]:
        # Manually create, start, set output, and end the span
        db_span = tracing.create_span("db_query", input={"query": query}, trace_id=trace_id)
        db_span.start() # Manually start the span
        try:
            time.sleep(random.uniform(0.1, 0.3)) # Simulate delay
            result = self._data.get(query, {})
            db_span.output = {"result": result}
            return result
        finally:
            db_span.end() # Manually end the span

def get_user_from_db_explicit(db: MockDatabase, user_id: int, trace_id: str) -> Dict[str, Any]:
    with tracing.create_span("get_user_from_db", input={"user_id": user_id}, trace_id=trace_id):
        query = f"SELECT * FROM users WHERE id = {user_id};"
        return db.execute_query(query, trace_id)

def main_explicit_control_example():
    db = MockDatabase()
    my_trace_id = str(uuid.uuid4())

    # Manually create a root span for the trace, providing the trace_id explicitly
    main_span = tracing.create_span("main_explicit_call", metadata={"env": "local"}, trace_id=my_trace_id)
    main_span.start() # Manually start the main span
    try:
        user = get_user_from_db_explicit(db, 1, my_trace_id)
        print(f"Retrieved user: {user.get('name')}")
    finally:
        main_span.end() # Manually end the main span

​
Exporting Historical or Pre-defined Tracing Data

You can pre-define start_time, end_time, span_id, parent_id, and trace_id when creating a span. This is useful for reporting historical data, replaying events, or reconstructing traces from external sources. After setting these attributes, call span.flush() to send the data.

import uuid
from datetime import datetime, timezone, timedelta
import scale_gp_beta.lib.tracing as tracing

parent_span_id = str(uuid.uuid4())
trace_id = str(uuid.uuid4())
child_span_id = str(uuid.uuid4())

now = datetime.now(timezone.utc)

# Parent Span: Create and configure a span with specific IDs and timestamps.
parent_span = tracing.create_span(
    "my_parent_span_name",
    input={"test": "input"},
    output={"test": "output"},
    metadata={"test": "metadata"},
    span_id=parent_span_id,
    trace_id=trace_id,
)
parent_span.start_time = (now - timedelta(minutes=10)).isoformat()
parent_span.end_time = now.isoformat()
parent_span.flush(blocking=True) # Send this span immediately

# Child Span: Link to the parent and trace using explicit IDs and timestamps.
child_span = tracing.create_span(
    "my_child_span_name",
    input={"test": "another input"},
    output={"test": "another output"},
    metadata={"test": "another metadata"},
    span_id=child_span_id,
    trace_id=trace_id,
    parent_id=parent_span_id, # Link to the parent span
)
child_span.start_time = (now - timedelta(minutes=6)).isoformat()
child_span.end_time = (now - timedelta(minutes=2)).isoformat()
child_span.flush() # Send this span immediately (defaults to blocking=True)

span.flush() by default blocks the main thread until the request has finished. For non-blocking behavior, use span.flush(blocking=False) to enqueue the request for the background worker. This is generally recommended for performance-sensitive applications.