sp-obs 0.1.1

Observability integration with Spinal

SP-OBS: Spinal OpenTelemetry Integration

SP-OBS is Spinal's cost tracking library built on top of open telemetry. It works by adding isolated tracers to libraries that have not been instrumented and attached a processor to libraries that aloready have been instrumented. This means we can also play nice with other observability libraries out there.

Features

• Seamlessly integrates with existing OpenTelemetry setups
• Works with Logfire, vanilla OpenTelemetry, or any OTEL-compatible framework
• Adds user and workflow context to spans for better tracking
• Selective span processing - only sends relevant AI/billing spans

Installation

pip install sp-obs

With AI Provider Support

# For OpenAI support
pip install sp-obs[openai]

# For Anthropic support  
pip install sp-obs[anthropic]

# For all providers
pip install sp-obs[all]

Quick Start

Configuration

First, configure SP-OBS with your endpoint and API key:

import sp_obs

# Configure globally (recommended)
sp_obs.configure(
    endpoint="https://your-endpoint.com/spans",
    api_key="your-api-key"
)

Or use environment variables:

• SPINAL_TRACING_ENDPOINT
• SPINAL_API_KEY

Instrumenting AI Providers

import sp_obs

# Configure SP-OBS
sp_obs.configure()

# Instrument providers
sp_obs.instrument_openai()
sp_obs.instrument_anthropic()
sp_obs.instrument_httpx()
sp_obs.instrument_requests()

Adding Context to Traces

Use the context manager to add user and workflow information:

import sp_obs

# Add context using context manager
with sp_obs.add_context(
    workflow_id="workflow-123",
    user_id="user-456",
    aggregation_id="session-789"  # optional
):
    # All spans created here will have this context
    response = client.chat.completions.create(...)

Configuration Options

Environment Variables

• SPINAL_TRACING_ENDPOINT: HTTP endpoint to send spans to
• SPINAL_API_KEY: API key for authentication
• SPINAL_PROCESS_MAX_QUEUE_SIZE: Max spans in queue (default: 2048)
• SPINAL_PROCESS_SCHEDULE_DELAY: Export delay in ms (default: 5000)
• SPINAL_PROCESS_MAX_EXPORT_BATCH_SIZE: Batch size (default: 512)
• SPINAL_PROCESS_EXPORT_TIMEOUT: Export timeout in ms (default: 30000)

Advanced Configuration

sp_obs.configure(
    endpoint="https://your-endpoint.com/spans",
    api_key="your-api-key",
    headers={"Custom-Header": "value"},
    timeout=5,
    max_queue_size=2048,
    max_export_batch_size=512,
    schedule_delay_millis=5000,
    export_timeout_millis=30000,
    scrubber=my_custom_scrubber  # Optional
)

Data Scrubbing

SP-OBS includes automatic scrubbing of sensitive data:

from sp_obs import DefaultScrubber, NoOpScrubber

# Use default scrubber (redacts tokens, keys, passwords)
sp_obs.configure(scrubber=DefaultScrubber())

# Or disable scrubbing
sp_obs.configure(scrubber=NoOpScrubber())

# Or implement custom scrubbing
class MyCustomScrubber:
    def scrub_attributes(self, attributes: dict) -> dict:
        # Your scrubbing logic
        return attributes

sp_obs.configure(scrubber=MyCustomScrubber())

Performance Considerations

SP-OBS uses a BatchSpanProcessor to minimize performance impact:

• Spans are batched and sent asynchronously in a background thread
• Default batch size: 512 spans
• Default flush interval: 5 seconds
• Spans are dropped if queue exceeds max size (default: 2048)

To tune for high-volume applications:

sp_obs.configure(
    max_queue_size=5000,          # Increase queue size
    max_export_batch_size=1000,   # Larger batches
    schedule_delay_millis=2000    # More frequent exports
)

What Spans Are Captured?

SP-OBS automatically captures:

• AI/LLM spans (identified by gen_ai.system attribute)
• HTTPX and request spans
• Explicitly created billing event spans
• Spans with attached user/workflow context

All other spans are ignored to minimize overhead and data transfer.

Integration Examples

FastAPI Application

from fastapi import FastAPI
import sp_obs
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

# Configure on startup
@app.on_event("startup")
async def startup():
    sp_obs.configure()
    sp_obs.instrument_openai()

@app.post("/generate")
async def generate(user_id: str, workflow_id: str):
    with sp_obs.add_context(user_id=user_id, workflow_id=workflow_id):
        response = await client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": "Hello"}]
        )
        return response

License

MIT License - see LICENSE file for details.