I've been working on AetherPhoto, a backend that organizes photos using natural language search and facial recognition. The architecture splits work across two main components: a FastAPI server that handles HTTP requests (like search and OAuth), and a Celery worker pool that processes the heavy lifting.
When you upload a batch of photos or sync from Google Photos, the API queues jobs in Redis. The Celery workers pick these up and run each image through a gauntlet: EXIF extraction, CLIP embedding for semantic search, zero-shot classification, perceptual hashing, and MTCNN/FaceNet for face detection. Finally, the data lands in a Neon PostgreSQL database and a Qdrant vector store.
graph LR
User[Browser] -->|HTTP| API[FastAPI API<br/>aetherphoto-api]
API -->|OTLP/gRPC| Collector[SigNoz OTel Collector<br/>:4317]
API -->|Dispatch Task| Redis[(Upstash Redis)]
Redis --> Worker[Celery Worker<br/>aetherphoto-worker]
Worker -->|OTLP/gRPC| Collector
API -->|SQL| PG[(PostgreSQL)]
Worker -->|SQL| PG
Worker -->|Vector Upsert/Search| Qdrant[(Qdrant)]
Worker -->|Download Photos| Google[Google Photos API]
Worker -->|CLIP + FaceNet| ML[ML Models]
Collector --> SigNoz[SigNoz UI<br/>:3301]
When an ingestion job took 40 seconds instead of four, my standard logging setup wasn't helping. I could see logs across the API and the worker, but stitching them together by timestamps to figure out if Qdrant was lagging or if the CLIP model inference was slow was miserable. I needed distributed tracing.
I decided to instrument the entire pipeline using OpenTelemetry and export the traces to a local instance of SigNoz. I chose SigNoz because it accepts OTLP directly and gave me a UI to query traces without shipping data to a managed service during local development.
The Telemetry Foundation
I started by creating a centralized telemetry.py module. Both the FastAPI app and the Celery workers need to initialize tracing, and I wanted them pointing to the same OTLP endpoint with consistent resource attributes.
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
_provider: TracerProvider | None = None
def init_telemetry(service_name: str) -> TracerProvider:
global _provider
if _provider is not None:
return _provider
resource = Resource.create(
{
SERVICE_NAME: service_name,
"deployment.environment": os.environ.get("ENV", "development"),
}
)
_provider = TracerProvider(resource=resource)
# Defaults to localhost:4317 for the SigNoz collector
otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
exporter = OTLPSpanExporter(
endpoint=otlp_endpoint,
insecure=not otlp_endpoint.startswith("https"),
)
span_processor = BatchSpanProcessor(exporter)
_provider.add_span_processor(span_processor)
trace.set_tracer_provider(_provider)
return _provider
I used BatchSpanProcessor so the tracer exports spans in the background without blocking the API or the ML worker threads. By calling init_telemetry("aetherphoto-api") in my main.py and init_telemetry("aetherphoto-worker") in celery_app.py, SigNoz immediately recognized both as distinct services.
Quick Wins: Auto-Instrumentation
OpenTelemetry's Python ecosystem includes auto-instrumentors that patch standard libraries at runtime. I added the instrumentors for FastAPI, SQLAlchemy, HTTPX, and Celery.
In main.py, the setup looks like this:
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from api.db.session import engine
# Trace outbound HTTP calls
HTTPXClientInstrumentor().instrument()
# Trace all DB queries
SQLAlchemyInstrumentor().instrument(engine=engine)
app = FastAPI(...)
# Trace incoming HTTP requests
FastAPIInstrumentor.instrument_app(app)
Hitting the /health endpoint immediately produced a trace. The FastAPIInstrumentor captured the route and HTTP method as the root span, and the SQLAlchemyInstrumentor captured the SELECT 1 query as a child span, including the exact SQL statement.
The Celery instrumentor is arguably the most valuable. By calling CeleryInstrumentor().instrument() in the worker setup, OpenTelemetry injects the current trace context into the Redis message headers when the API calls .delay(). When the worker pulls the task, it extracts those headers and continues the trace.
Filling the Gaps: Manual Spans
Auto-instrumentation handles HTTP and SQL, but it doesn't know what my ML pipeline is doing. Inside embed_photo_task.py, a single Celery task runs image downloads, EXIF extraction, CLIP embedding, and face detection sequentially. Without manual spans, this entire process would show up in SigNoz as one massive, opaque block of execution time.
I needed to see exactly how long the CLIP inference took compared to the Qdrant vector upsert. I imported get_tracer(__name__) and wrapped each distinct operation using tracer.start_as_current_span().
Here is how I instrumented the Qdrant and zero-shot classification steps:
# 3. Store Embedding in Qdrant Vector database
with tracer.start_as_current_span(
"qdrant.upsert_embedding",
attributes={
"db.system": "qdrant",
"db.operation": "upsert",
"photo.id": photo_id,
},
) as span:
vector_store.upsert_image_embedding(photo.id, embedding, payload)
# 4. Zero-Shot Classification
with tracer.start_as_current_span(
"clip.zero_shot_classify",
attributes={
"photo.id": photo_id,
"clip.model": "openai/clip-vit-base-patch32",
},
) as span:
category = category_service.classify_image(embedding)
photo.category = category
span.set_attribute("photo.category", category)
Adding custom attributes like photo.category and photo.id proved useful. I can now filter in SigNoz to find all traces where the CLIP model classified an image as "documents" to see if those take longer to process than "landscapes".
I applied this same pattern across the codebase: wrapping the DBSCAN face clustering, the perceptual hash duplicate detection, and the Google Photos Picker polling loop.
Catching Hidden Failures
My Google Photos ingestion flow relies on fetching temporary URLs from the Picker API. These URLs are ephemeral and expire after about an hour. If a user selects 5,000 photos, the background ingestion might take longer than an hour, meaning the last few hundred image downloads will fail.
Before tracing, I'd just see a generic task failure in the Celery logs. With my manual spans recording exceptions, the failure mode is obvious. I catch the exception inside the image.download span, call span.record_exception(e), and set the span status to error:
with tracer.start_as_current_span("image.download") as span:
try:
response = httpx.get(photo.file_path_or_url)
response.raise_for_status()
# ... write to file ...
except Exception as e:
span.record_exception(e)
span.set_status(trace.StatusCode.ERROR, str(e))
raise
When the token expires, the HTTPX client throws an HTTPStatusError for a 403 Forbidden. The root embed_photo_task span is marked as an error, and the exact child span (image.download) holds the 403 response text. I can filter my traces by photo.source="google_photos" and status=ERROR to immediately see how many photos failed due to token expiration.
Mapping the Topology
Because I passed engine to the SQLAlchemyInstrumentor and wrapped Qdrant calls with attributes like db.system: qdrant, SigNoz automatically inferred the topology of my application.
The Service Map view visualizes this beautifully. I can see the API server dispatching messages to Redis, the worker consuming them, and then the worker fanning out connections to PostgreSQL, Qdrant, and external Google domains. When I deploy this to production, this map will give me an immediate visual indicator of where latency is pooling—whether it's the database queries slowing down or the vector store struggling under load.
Top comments (0)