# End-to-End Tracing? Autotel Makes It Automatic

By Jag Reehal, 2026-01-15. Canonical: https://arrangeactassert.com/posts/oso-end-to-end-tracing/

<!-- Excerpt Start -->

The OSO team is spot on in their [End-to-End Tracing in Event Driven Architectures](https://oso.sh/blog/end-to-end-tracing-in-event-driven-architectures/) post.

> Traces break at queues unless you extract context from message headers and put it in the appropriate context.

They walk through the real pain: stateful processing loses trace context in caches, Kafka Connect can only do batch-level tracing, and every team ends up writing custom interceptors and state store wrappers.

We've all been there.

Their key insight:

> In Kafka Streams and Kafka Connect this often means manual work: interceptors, state stores, batch spans, or extending tracing logic to extract from headers.

But there's still a gap...

<!-- Excerpt End -->

I built Autotel to close that gap.

Autotel does not change Kafka semantics or tracing theory; it standardizes how OpenTelemetry context propagation is applied across services.

Autotel makes end-to-end tracing the default. No custom interceptors. No state store wrappers. No per-service boilerplate for inject/extract.

## The Problem You're Solving

A checkout request flows through your event-driven system:

```text
API → Kafka → payment-worker → Kafka → email-worker → Kafka → batcher
```

Without proper context propagation, you get:

```text
Trace A (API)
Trace B (payment-worker)  ← new trace, no link to A
Trace C (email-worker)    ← new trace, no link to A or B
Trace D (batcher)         ← batch span, no idea which requests contributed
```

You cannot see the full path in Jaeger. You cannot query "everything for this checkout." When the batcher fails, you have no idea which upstream requests were affected.

The manual fix? Every team writes:

```ts
// Manual approach: custom interceptor
class TracingProducerInterceptor implements ProducerInterceptor {
  onSend(record: ProducerRecord): ProducerRecord {
    const span = tracer.startSpan('kafka.send', { kind: SpanKind.PRODUCER });
    const headers = record.headers ?? {};
    propagator.inject(context.active(), headers, {
      set: (carrier, key, value) => { carrier[key] = value; }
    });
    record.headers = headers;
    span.setAttribute('messaging.system', 'kafka');
    span.setAttribute('messaging.destination.name', record.topic);
    span.end();
    return record;
  }
}

// Then a consumer interceptor...
class TracingConsumerInterceptor implements ConsumerInterceptor {
  onConsume(record: ConsumerRecord): ConsumerRecord {
    const parentContext = propagator.extract(context.active(), record.headers, {
      get: (carrier, key) => carrier[key]?.toString()
    });
    return context.with(parentContext, () => {
      const span = tracer.startSpan('kafka.process', { kind: SpanKind.CONSUMER });
      span.setAttribute('messaging.system', 'kafka');
      span.setAttribute('messaging.kafka.partition', record.partition);
      span.setAttribute('messaging.kafka.offset', record.offset);
      // ... more attributes
      return record;
    });
  }
}

// And for batches, custom state stores, batch spans with links...
```

It works, but it's a lot of ceremony for something we all want to be the default.

## The Solution: Zero Per-Service Wiring

With Autotel, the same flow uses `withProducerSpan()` and `withConsumerSpan()`:

```ts
// Producer: API sends to Kafka
import { withProducerSpan, injectTraceHeaders } from "@demo/observability/kafka";

async function sendEvent(payload: Envelope, headers: Record<string, string>) {
  await withProducerSpan(
    "orders",
    { messageKey: payload.correlation_id },
    async () => {
      const h = injectTraceHeaders(headers);
      await producer.send({
        messages: [{
          topic: "orders",
          key: payload.correlation_id,
          value: JSON.stringify(payload),
          headers: h,  // traceparent, tracestate, x-correlation-id
        }],
      });
    },
  );
}
```

```ts
// Consumer: Worker processes from Kafka
import { withConsumerSpan, normalizeHeaders } from "@demo/observability/kafka";

for await (const msg of stream) {
  const headers = normalizeHeaders(msg.headers);
  const messageMeta = { partition: msg.partition, offset: String(msg.offset) };

  await withConsumerSpan(
    "orders",
    "email-worker",
    headers,
    messageMeta,
    async () => {
      // Your business logic here
      // Context is already extracted - you're in the same trace
      await handleOrderCreated(msg);
    },
  );
}
```

No custom interceptors. No `propagator.inject()` / `propagator.extract()`. No manual span attribute setting.

### Batch Lineage for Fan-In

When multiple messages feed into one batch, Autotel's `extractBatchLineage()` creates a batch span with links:

```ts
import { extractBatchLineage, withProcessingSpan, SEMATTRS_LINKED_TRACE_ID_COUNT, SEMATTRS_LINKED_TRACE_ID_HASH } from "@demo/observability/kafka";

async function doFlush() {
  const items = batch;
  batch = [];

  // Extract lineage from all contributing messages
  const lineage = extractBatchLineage(
    items.map((it) => ({ headers: it.headers })),
    { maxLinks: 50 },
  );

  await withProcessingSpan(
    {
      name: "v1.settlements.batched process",
      headers: {},
      contextMode: "none",
      links: lineage.links,  // Links to contributing traces
      topic: "settlements",
      consumerGroup: "batcher",
    },
    async (span) => {
      span.setAttribute(SEMATTRS_LINKED_TRACE_ID_COUNT, lineage.linked_trace_id_count);
      span.setAttribute(SEMATTRS_LINKED_TRACE_ID_HASH, lineage.linked_trace_id_hash);
      await processBatch(items);
    },
  );
}
```

The batch span carries:
- `linked_trace_id_count`: How many traces contributed
- `linked_trace_id_hash`: Stable hash for querying "same set of inputs"
- `links`: Actual trace links to contributing spans (capped to avoid payload explosion)

## Manual Interceptors vs Autotel

| Manual Interceptors | Autotel |
| --- | --- |
| Custom ProducerInterceptor class | `withProducerSpan()` wrapper |
| Custom ConsumerInterceptor class | `withConsumerSpan()` wrapper |
| Manual `propagator.inject()` | `injectTraceHeaders()` |
| Manual `propagator.extract()` | Automatic in `withConsumerSpan()` |
| Per-attribute `span.setAttribute()` | Semantic conventions automatic |
| Custom state store wrappers for stateful processing | Same pattern, context flows through stateful processing |
| Manual batch spans with child spans | `extractBatchLineage()` with links |
| Different code per broker | Same API (`withProducerSpan` / `withConsumerSpan`) |

## One Trace, API to Batcher

With Autotel, the same checkout request shows as one trace in Jaeger:

```text
Trace: 4bf92f3577b34da6a3ce929d0e0e4736
├─ api: POST /checkout (INTERNAL)
│  └─ orders publish (PRODUCER)
├─ orders process (CONSUMER) - payment-worker
│  └─ payments publish (PRODUCER)
├─ payments process (CONSUMER) - email-worker
│  └─ notifications publish (PRODUCER)
└─ v1.settlements.batched process (CONSUMER) - batcher
   └─ linked_trace_id_count: 5
   └─ linked_trace_id_hash: a3f2c1...
```

Query `correlation_id` to find all related logs, events, and traces. Query `linked_trace_id_hash` to find all batches with the same contributing requests. Retries and dead-letter forwarding preserve `correlation_id` and trace context so failures remain observable across hops.

## Try It Yourself

Clone the demo and see end-to-end traces in Jaeger:

```bash
git clone https://github.com/jagreehal/event-driven-observability-demo
cd event-driven-observability-demo
pnpm install
docker compose up -d  # Kafka, Jaeger
pnpm build && pnpm start
```

Trigger a checkout:

```bash
curl -sX POST http://localhost:3000/rpc/checkout/start \
  -H "content-type: application/json" \
  -d '{"cartValue":149.99,"itemCount":2,"userId":"usr_123","tenantId":"acme"}' | jq
```

Open Jaeger at `http://localhost:16686`. Search for the trace. You'll see one connected trace from API through workers, including batch spans with lineage.

### Evidence from a live demo run

Here's actual output from a demo run (2026-02-07):

**Jaeger trace** (queried via Jaeger API):

```json
{
  "traceID": "429a7e3408f6408f259348df534db2fc",
  "spanCount": 3,
  "services": ["api", "worker-email"],
  "spans": [
    {
      "operation": "checkout/start",
      "kind": "internal",
      "service": "api"
    },
    {
      "operation": "domain-events publish",
      "kind": "producer",
      "attributes": {
        "messaging.system": "kafka",
        "messaging.destination.name": "domain-events",
        "messaging.operation": "publish",
        "messaging.kafka.message.key": "429a7e3408f6408f"
      }
    },
    {
      "operation": "domain-events process",
      "kind": "consumer",
      "attributes": {
        "messaging.system": "kafka",
        "messaging.destination.name": "domain-events",
        "messaging.kafka.consumer.group": "demo-v1",
        "messaging.kafka.partition": 2,
        "messaging.kafka.offset": "0"
      }
    }
  ]
}
```

One trace (`429a7e3408f6408f...`) spans `api` and `worker-email`. The CONSUMER span's parent is the PRODUCER span — no custom interceptors, no state store wrappers, just `withProducerSpan` and `withConsumerSpan`.

Repo: [github.com/jagreehal/event-driven-observability-demo](https://github.com/jagreehal/event-driven-observability-demo)