Observability Audit & Improvement Plan

May contain stale references

This document was audited on 2026-06-22 and contains references that need updating: 3-service claim is historical; correlation.ts now exists. Cross-check critical details against the current codebase before relying on them.

Executive Summary

This document provides a comprehensive audit of the observability implementation in the ERP-Unlocked monorepo, identifies gaps, and proposes a unified tracing solution to enable end-to-end request tracking across all services.

Key Findings

• 3 services have full observability coverage (webapp, pdf-api, pdf-worker)
• 3 services have partial observability coverage (dagster, trigger, email-api)
• 4 observability platforms are in use (Better Stack, Langfuse, OpenTelemetry, Logtail)
• Critical gap: Traces are fragmented across services, making end-to-end debugging difficult

Proposed Solution

Implement a Correlation ID + Unified OTEL approach that enables:

• Single trace from user action → final result
• Cross-service trace propagation (including Trigger.dev and Dagster)
• Unified querying in Better Stack by correlation_id

Table of Contents

Part 1: Current State Analysis

1. Observability Tools Overview
2. Tool Responsibilities
3. Service-by-Service Analysis
4. Current Trace Flow
5. Implementation Gaps

Part 2: Proposed Solution

6. Unified Tracing Architecture
7. Implementation Plan
8. Code Examples

Appendix

9. Reference

Part 1: Current State Analysis

1. Observability Tools Overview

1.1 Tools Ecosystem

flowchart TB
    subgraph Platforms["☁️ Observability Platforms"]
        BS["Better Stack<br/>───────────<br/>Logs, Traces, Metrics"]
        LF["Langfuse<br/>───────────<br/>LLM Observability"]
    end

    subgraph Standards["📐 Standards & Protocols"]
        OTEL["OpenTelemetry<br/>───────────<br/>OTLP Protocol"]
        W3C["W3C Trace Context<br/>───────────<br/>traceparent header"]
    end

    subgraph NodeLibs["📦 Node.js Libraries"]
        OtelNode["@opentelemetry/sdk-node"]
        Pino["Pino Logger"]
        RepoObs["@repo/observability"]
    end

    subgraph PyLibs["🐍 Python Libraries"]
        OtelPy["opentelemetry-python"]
        Structlog["structlog"]
        Logtail["logtail-python"]
    end

    RepoObs --> OtelNode
    RepoObs --> Pino
    OtelNode --> OTEL
    Pino --> OTEL

    OtelPy --> OTEL
    Structlog --> OTEL
    Logtail --> BS

    OTEL --> BS
    OTEL --> W3C
    LF -.->|"Isolated tracer"| OTEL

    style Platforms fill:#60a5fa,color:#000
    style Standards fill:#a78bfa,color:#000
    style NodeLibs fill:#4ade80,color:#000
    style PyLibs fill:#fbbf24,color:#000

1.2 Tools Summary

Tool	Type	Primary Use	Services
OpenTelemetry	SDK/Protocol	Distributed tracing, metrics	webapp, pdf-api, pdf-worker
Better Stack	Platform	Log aggregation, traces	All instrumented services
Langfuse	Platform	LLM observability	pdf-worker
Pino	Library (Node.js)	Structured logging	webapp, trigger, email-api
structlog	Library (Python)	Structured logging	pdf-api, pdf-worker

1.3 Platform Endpoints

Platform	Endpoint	Protocol
Better Stack (OTLP)	https://in-otel.betterstack.com	HTTP/Protobuf
Better Stack (Logs)	https://in.logs.betterstack.com	Logtail SDK
Langfuse	https://cloud.langfuse.com	REST API

2. Tool Responsibilities

2.1 OpenTelemetry (OTEL)

Package: packages/observability/

Responsibility	Implementation
Distributed Tracing	Auto span creation, W3C traceparent propagation
Metrics Collection	HTTP counters, DB duration, sync health scores
Log Correlation	trace_id/span_id injection via Pino logHook

Key Files:

• packages/observability/src/otel-config.ts - NodeSDK initialization
• packages/observability/src/trace.ts - withTrace, setTraceAttribute
• apps/webapp/otel-init.cjs - Webapp bootstrap

2.2 Better Stack

Responsibility	Features
Log Management	Centralized aggregation, full-text search
Trace Visualization	Distributed trace viewing, latency analysis
Alerting	Error rate, latency threshold alerts

2.3 Langfuse

Package: packages/pdf-shared/pdf_shared/utils/llm_observability.py

Responsibility	Features
LLM Tracking	Gemini API monitoring, duration tracking
Token Usage	Input/output counting, aggregation
Cost Calculation	$0.075/1M input, $0.30/1M output tokens
Quality Scoring	JSON parsing, schema validation, extraction success

3. Service-by-Service Analysis

3.1 Coverage Summary

flowchart LR
    subgraph Full["✅ FULL COVERAGE"]
        W["webapp"]
        PA["pdf-api"]
        PW["pdf-worker"]
    end

    subgraph Partial["⚠️ PARTIAL COVERAGE"]
        D["dagster"]
        T["trigger"]
        E["email-api"]
    end

    style Full fill:#4ade80,color:#000
    style Partial fill:#fbbf24,color:#000

3.2 Detailed Analysis

Service	Tracing	Metrics	Logging	LLM Obs	Status
webapp	✅	✅	✅	N/A	Full OTEL via otel-init.cjs
pdf-api	✅	✅	✅	N/A	Python OTEL auto-instrumentation
pdf-worker	✅	✅	✅	✅	Full OTEL + Langfuse
dagster	⚠️	❌	⚠️	N/A	Service name only, no SDK
trigger	❌	❌	✅	N/A	Native logger, no OTEL
email-api	❌	❌	✅	N/A	Pino only, CF Worker limits

4. Current Trace Flow

4.1 What Works: PDF Processing

sequenceDiagram
    autonumber
    participant User as 👤 User
    participant Webapp as 🌐 Webapp
    participant BS as ☁️ Better Stack
    participant PdfApi as ⚡ PDF-API
    participant PdfWorker as ⚙️ PDF-Worker
    participant LF as 🤖 Langfuse

    User->>Webapp: Upload PDF
    Note over Webapp: trace_id: abc123
    Webapp->>BS: Span + Logs

    Webapp->>PdfApi: HTTP + traceparent
    Note over PdfApi: trace_id: abc123 (continued)
    PdfApi->>BS: Span + Logs

    PdfApi->>PdfWorker: Celery + context
    Note over PdfWorker: trace_id: abc123 (continued)
    PdfWorker->>LF: LLM spans
    PdfWorker->>BS: Span + Logs

4.2 What’s Broken: Trigger.dev & Dagster

sequenceDiagram
    autonumber
    participant User as 👤 User
    participant Webapp as 🌐 Webapp
    participant Trigger as ⏰ Trigger.dev
    participant Dagster as 📊 Dagster
    participant BS as ☁️ Better Stack

    User->>Webapp: Start ERP Sync
    Note over Webapp: trace_id: abc123
    Webapp->>BS: Span

    Webapp->>Trigger: Trigger task (NO trace context!)
    Note over Trigger: ❌ NEW trace_id: xyz789
    Trigger->>BS: Separate trace

    Trigger->>Dagster: Call Dagster
    Note over Dagster: ❌ NO tracing at all

Problem: Cannot follow a single request across Trigger.dev and Dagster jobs.

5. Implementation Gaps

5.1 Gap Summary

flowchart LR
    subgraph Critical["🔴 CRITICAL"]
        G1["Fragmented Traces<br/>Cannot follow requests E2E"]
        G2["Dagster Invisible<br/>No OTEL SDK"]
    end

    subgraph Moderate["🟡 MODERATE"]
        G3["Trigger.dev Isolated<br/>No trace propagation"]
        G4["Pino Correlation<br/>trace_id may be missing"]
    end

    style Critical fill:#f87171,color:#000
    style Moderate fill:#fbbf24,color:#000

5.2 Gap Details

Gap	Impact	Current State
Fragmented Traces	Cannot debug cross-service issues	Each service starts new trace
Dagster No OTEL	ETL jobs invisible	Only service name configured
Trigger.dev Isolated	Background jobs disconnected	Native logger only
Langfuse Separate	LLM traces not linked	Isolated tracer provider

Part 2: Proposed Solution

6. Unified Tracing Architecture

6.1 Current vs Proposed

flowchart TB
    subgraph Current["❌ CURRENT - Fragmented"]
        direction LR
        C1[Webapp Trace 1] --> C2[PDF-API Trace 1]
        C3[Trigger Trace 2] --> C4[Dagster No Trace]
    end

    subgraph Proposed["✅ PROPOSED - Unified"]
        direction LR
        P1[Webapp] --> P2[PDF-API]
        P1 --> P3[Trigger]
        P3 --> P4[Dagster]
        P5[correlation_id: abc123]
    end

    style Current fill:#fee2e2
    style Proposed fill:#dcfce7

6.2 Proposed Architecture

sequenceDiagram
    autonumber
    participant U as User
    participant W as Webapp
    participant T as Trigger.dev
    participant D as Dagster
    participant PA as PDF-API
    participant PW as PDF-Worker
    participant BS as Better Stack

    U->>W: Request
    Note over W: Generate correlation_id: abc123

    W->>BS: Span (correlation_id: abc123)

    W->>T: Task + _traceContext {correlation_id, traceparent}
    Note over T: Restore parent trace
    T->>BS: Child span (correlation_id: abc123)

    T->>D: Job + correlation_id
    Note over D: New OTEL span with correlation_id
    D->>BS: Span (correlation_id: abc123)

    W->>PA: HTTP + traceparent + x-correlation-id
    PA->>PW: Celery + context
    PW->>BS: All spans (correlation_id: abc123)

    Note over BS: Query by correlation_id = abc123<br/>See ALL spans across ALL services

6.3 Key Concept: Correlation ID

A correlation_id is a UUID generated at the entry point (webapp) that is:

• Passed in HTTP headers (x-correlation-id)
• Included in Trigger.dev task payloads
• Set as span attribute in all services
• Used as Langfuse session_id
• Queryable in Better Stack

7. Implementation Plan

7.1 Timeline

Phase	Task	Effort	Priority
1	Correlation ID infrastructure	2-3 hrs	🔴 High
2	Webapp integration	1-2 hrs	🔴 High
3	Trigger.dev integration	3-4 hrs	🔴 High
4	Dagster OTEL	2-3 hrs	🔴 High
5	Langfuse correlation	1 hr	🟡 Medium
6	Better Stack queries	30 min	🟢 Low

Total: 10-14 hours

7.2 Quick Win Option (4 hours)

If full implementation is too much, start with correlation ID only:

1. ✅ Generate correlation_id in webapp middleware
2. ✅ Pass in HTTP headers and Trigger payloads
3. ✅ Log in all services
4. ✅ Query Better Stack by correlation_id

Result: 80% benefit with 30% effort.

7.3 Files to Create/Modify

New Files:

packages/observability/src/correlation.ts
apps/trigger/src/utils/tracing.ts
apps/dagster/erp_pipeline/observability.py

Modified Files:

apps/webapp/src/middleware.ts
apps/webapp/src/lib/observability/fetchWithTrace.ts
apps/trigger/src/tasks/**/*.ts
packages/pdf-shared/pdf_shared/utils/llm_observability.py

8. Code Examples

8.1 Correlation ID Infrastructure

packages/observability/src/correlation.ts

import { randomUUID } from 'crypto';
import { context, propagation } from '@opentelemetry/api';

export const CORRELATION_ID_HEADER = 'x-correlation-id';
export const CORRELATION_ID_BAGGAGE_KEY = 'correlation.id';

export function getOrCreateCorrelationId(existingId?: string): string {
  return existingId || randomUUID();
}

export function extractCorrelationId(headers: Headers): string | undefined {
  return headers.get(CORRELATION_ID_HEADER) || undefined;
}

export function withCorrelationId<T>(correlationId: string, fn: () => T): T {
  const baggage = propagation.createBaggage({
    [CORRELATION_ID_BAGGAGE_KEY]: { value: correlationId },
  });
  const ctx = propagation.setBaggage(context.active(), baggage);
  return context.with(ctx, fn);
}

8.2 Webapp Middleware

apps/webapp/src/middleware.ts

import {
  getOrCreateCorrelationId,
  extractCorrelationId,
  withCorrelationId,
  setCorrelationIdAttribute,
  CORRELATION_ID_HEADER,
} from '@repo/observability';
import { setTraceAttribute } from '@repo/observability';

export const onRequest = defineMiddleware(async (context, next) => {
  const incomingId = extractCorrelationId(context.request.headers);
  const correlationId = getOrCreateCorrelationId(incomingId);

  // Set as trace attribute
  setTraceAttribute('correlation.id', correlationId);

  // Store for use in routes
  context.locals.correlationId = correlationId;

  const response = await next();
  response.headers.set('x-correlation-id', correlationId);
  return response;
});

8.3 Trigger.dev Task with Trace Restoration

apps/trigger/src/utils/tracing.ts

import { trace, context, propagation, SpanKind } from '@opentelemetry/api';

const tracer = trace.getTracer('trigger-tasks');

export interface TraceablePayload {
  _traceContext?: {
    correlationId: string;
    traceparent?: string;
  };
}

export async function withRestoredTrace<T>(
  taskName: string,
  payload: TraceablePayload,
  fn: () => Promise<T>
): Promise<T> {
  const traceContext = payload._traceContext;

  if (!traceContext?.traceparent) {
    return tracer.startActiveSpan(taskName, fn);
  }

  // Restore parent context
  const parentContext = propagation.extract(context.active(), {
    traceparent: traceContext.traceparent,
  });

  return context.with(parentContext, () => {
    return tracer.startActiveSpan(taskName, { kind: SpanKind.CONSUMER }, async span => {
      span.setAttribute('correlation.id', traceContext.correlationId);
      try {
        return await fn();
      } finally {
        span.end();
      }
    });
  });
}

8.4 Dagster OTEL Setup

apps/dagster/erp_pipeline/observability.py

import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from functools import wraps

def setup_observability():
    resource = Resource.create({
        "service.name": "dagster",
        "service.version": "1.0.0",
    })

    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(provider)

    return trace.get_tracer(__name__)

tracer = setup_observability()

def traced_asset(name: str):
    """Decorator for tracing Dagster assets"""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            correlation_id = kwargs.get('correlation_id')
            with tracer.start_as_current_span(name) as span:
                if correlation_id:
                    span.set_attribute("correlation.id", correlation_id)
                return fn(*args, **kwargs)
        return wrapper
    return decorator

8.5 Triggering Tasks with Context

apps/webapp/src/services/trigger-service.ts

import { tasks } from '@trigger.dev/sdk/v3';
import { context, propagation } from '@opentelemetry/api';

export async function triggerTaskWithTrace(
  taskId: string,
  payload: Record<string, unknown>,
  correlationId: string
) {
  // Capture current trace context
  const carrier: Record<string, string> = {};
  propagation.inject(context.active(), carrier);

  return tasks.trigger(taskId, {
    ...payload,
    _traceContext: {
      correlationId,
      traceparent: carrier.traceparent,
    },
  });
}

9. Reference

9.1 Metrics Reference

Category	Metric	Type
HTTP	httpRequestsTotal	Counter
HTTP	httpRequestDuration	Histogram
Database	dbQueryDuration	Histogram
ERP	erpRequestsTotal	Counter
LLM	llm_tokens_total	Counter
LLM	llm_request_cost_usd	Counter

9.2 Related Documentation

• docs/observability/README.md - Overview
• docs/observability/better-stack-integration-guide.md - Better Stack setup
• packages/observability/README.md - Package docs
• learnings/DEBUGGING_LEARNINGS.md - Past debugging