E2B Sandbox Monitoring & Tracing with OpenTelemetry

import logging
import os
 
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
resource = Resource.create(
    {
        "e2b.sandbox.id": os.environ.get("E2B_SANDBOX_ID", ""),
        "e2b.template.id": os.environ.get("E2B_TEMPLATE_ID", ""),
    }
)
 
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(tracer_provider)
 
logger_provider = LoggerProvider(resource=resource)
logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter()))
logging.getLogger().addHandler(LoggingHandler(logger_provider=logger_provider))
logging.getLogger().setLevel(logging.INFO)
 
# The parent context arrives as an environment variable. The OpenTelemetry SDK
# does not read TRACEPARENT on its own, so extract it here.
parent = TraceContextTextMapPropagator().extract(
    {"traceparent": os.environ.get("TRACEPARENT", "")}
)
 
tracer = trace.get_tracer("sandbox-workload")
log = logging.getLogger("sandbox-workload")
 
with tracer.start_as_current_span("workload", context=parent) as span:
    log.info("starting work")
    with tracer.start_as_current_span("workload.compute"):
        total = sum(i * i for i in range(200000))
    log.info("compute finished with total %s", total)
    span.set_attribute("workload.total", total)
 
tracer_provider.force_flush()
logger_provider.force_flush()

import os
 
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
from telemetry import tracer
from traced_sandbox import traced_command, traced_sandbox
 
with traced_sandbox(timeout=300) as sandbox:
    with open("sandbox_app.py", "rb") as f:
        sandbox.files.write("/home/user/sandbox_app.py", f.read())
 
    traced_command(
        sandbox,
        "pip install --quiet opentelemetry-sdk opentelemetry-exporter-otlp-proto-http",
    )
 
    carrier = {}
    TraceContextTextMapPropagator().inject(carrier)
 
    traced_command(
        sandbox,
        "python3 /home/user/sandbox_app.py",
        envs={
            "TRACEPARENT": carrier["traceparent"],
            "OTEL_SERVICE_NAME": "<your-sandbox-service-name>",
            # Forward the two variables Step 2 already exported, so the
            # endpoint and the ingestion key are defined in exactly one place.
            "OTEL_EXPORTER_OTLP_ENDPOINT": os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"],
            "OTEL_EXPORTER_OTLP_HEADERS": os.environ["OTEL_EXPORTER_OTLP_HEADERS"],
        },
    )

import time
 
from opentelemetry import metrics
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
 
reader = PeriodicExportingMetricReader(OTLPMetricExporter())
meter_provider = MeterProvider(resource=Resource.create(), metric_readers=[reader])
metrics.set_meter_provider(meter_provider)
meter = metrics.get_meter("e2b-orchestrator")
 
# Sandboxes to report on, keyed by sandbox id. Add on create, remove on kill.
tracked = {}
 
# Every gauge below runs its own callback, and they all fire within the same
# collection. Without this cache that is one get_metrics() call per gauge, so
# six per sandbox per collection. The TTL is long enough to cover one
# collection and shorter than any sane export interval, so each collection
# still reads fresh values.
_CACHE_TTL_SECONDS = 2.0
_cache = {}
 
 
def _latest(sandbox_id, sandbox):
    now = time.monotonic()
    cached = _cache.get(sandbox_id)
    if cached is not None and now - cached[0] < _CACHE_TTL_SECONDS:
        return cached[1]
 
    samples = sandbox.get_metrics()
    sample = samples[-1] if samples else None
    _cache[sandbox_id] = (now, sample)
    return sample
 
 
def _observe(field, scale=1):
    def callback(options):
        for sandbox_id, sandbox in list(tracked.items()):
            sample = _latest(sandbox_id, sandbox)
            if sample is None:
                continue
            yield metrics.Observation(
                getattr(sample, field) * scale,
                {"e2b.sandbox.id": sandbox_id},
            )
 
    return callback
 
 
meter.create_observable_gauge(
    "e2b.sandbox.cpu.used_pct", callbacks=[_observe("cpu_used_pct")], unit="%"
)
meter.create_observable_gauge(
    "e2b.sandbox.cpu.count", callbacks=[_observe("cpu_count")], unit="{cpu}"
)
meter.create_observable_gauge(
    "e2b.sandbox.memory.used", callbacks=[_observe("mem_used")], unit="By"
)
meter.create_observable_gauge(
    "e2b.sandbox.memory.total", callbacks=[_observe("mem_total")], unit="By"
)
meter.create_observable_gauge(
    "e2b.sandbox.disk.used", callbacks=[_observe("disk_used")], unit="By"
)
meter.create_observable_gauge(
    "e2b.sandbox.disk.total", callbacks=[_observe("disk_total")], unit="By"
)

import os
import time
 
import requests
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.sdk.resources import Resource
import logging
 
logger_provider = LoggerProvider(resource=Resource.create())
logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter()))
log = logging.getLogger("e2b-lifecycle")
log.addHandler(LoggingHandler(logger_provider=logger_provider))
log.setLevel(logging.INFO)
 
seen = set()
 
 
def poll_once():
    response = requests.get(
        "https://api.e2b.app/events/sandboxes",
        params={"limit": 100},
        headers={"X-API-Key": os.environ["E2B_API_KEY"]},
        timeout=30,
    )
    response.raise_for_status()
 
    for event in response.json():
        if event["id"] in seen:
            continue
        seen.add(event["id"])
        log.info(
            event["type"],
            extra={
                "e2b.sandbox.id": event["sandboxId"],
                "e2b.template.id": event["sandboxTemplateId"],
                "e2b.event.type": event["type"],
                "e2b.event.timestamp": event["timestamp"],
            },
        )
 
 
while True:
    poll_once()
    logger_provider.force_flush()
    time.sleep(30)

span.set_attributes(
    {
        "gen_ai.operation.name": "execute_tool",
        "gen_ai.tool.name": "e2b_sandbox",
        "gen_ai.tool.type": "extension",
        "gen_ai.tool.call.id": tool_call.id,
    }
)