Request ID Tracing: Track Requests Across Services with Correlation IDs

Problem

In distributed systems, a single user request often traverses multiple microservices. Without proper tracking:

• Debugging is nightmarish: When an error occurs, you can't trace which service caused it
• Performance issues are invisible: You can't identify bottlenecks across the call chain
• Audit trails are incomplete: Compliance requirements go unmet
• Root cause analysis fails: Multiple services log independently with no connection
• Monitoring becomes fragmented: Each service has isolated metrics

Example: User clicks "checkout" → Payment Service → Inventory Service → Notification Service → Database. If something fails, which service is responsible?

Solution

Correlation IDs (also called Request IDs or Trace IDs) are unique identifiers that follow a request through its entire lifecycle across all services. Each service:

1. Receives or generates a correlation ID
2. Includes it in all logs
3. Passes it to downstream services
4. Returns it in responses

This creates an observable request chain enabling end-to-end tracing.

---

Code Implementation

1. Node.js/Express with Correlation IDs

// middleware/correlationId.js
const { v4: uuidv4 } = require('uuid');

const correlationIdMiddleware = (req, res, next) => {
  // Use existing correlation ID from header or generate new one
  const correlationId = req.headers['x-correlation-id'] || uuidv4();

  // Store in request object
  req.correlationId = correlationId;

  // Add to response headers
  res.setHeader('x-correlation-id', correlationId);

  // Store in async local storage for access in nested calls
  asyncLocalStorage.run(correlationId, () => {
    next();
  });
};

module.exports = correlationIdMiddleware;

// utils/asyncLocalStorage.js
const { AsyncLocalStorage } = require('async_hooks');

const asyncLocalStorage = new AsyncLocalStorage();

const getCorrelationId = () => asyncLocalStorage.getStore();

module.exports = { asyncLocalStorage, getCorrelationId };

// logger/logger.js
const winston = require('winston');
const { getCorrelationId } = require('../utils/asyncLocalStorage');

const logger = winston.createLogger({
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.printf(({ timestamp, level, message }) => {
      const correlationId = getCorrelationId();
      return `[${timestamp}] [${correlationId}] [${level}]: ${message}`;
    })
  ),
  transports: [
    new winston.transports.Console(),
    new winston.transports.File({ filename: 'app.log' })
  ]
});

module.exports = logger;

// services/paymentService.js
const axios = require('axios');
const { getCorrelationId } = require('../utils/asyncLocalStorage');
const logger = require('../logger/logger');

async function processPayment(orderId, amount) {
  const correlationId = getCorrelationId();

  logger.info(`Processing payment for order ${orderId}`);

  try {
    // Call inventory service with correlation ID
    const response = await axios.post(
      'http://inventory-service/api/reserve',
      { orderId, quantity: 1 },
      {
        headers: {
          'x-correlation-id': correlationId,
          'Content-Type': 'application/json'
        }
      }
    );

    logger.info(`Inventory reserved successfully`);
    return response.data;
  } catch (error) {
    logger.error(`Payment processing failed: ${error.message}`);
    throw error;
  }
}

module.exports = { processPayment };

// app.js
const express = require('express');
const correlationIdMiddleware = require('./middleware/correlationId');
const logger = require('./logger/logger');
const { processPayment } = require('./services/paymentService');

const app = express();

// Apply correlation ID middleware
app.use(correlationIdMiddleware);
app.use(express.json());

app.post('/api/checkout', async (req, res) => {
  try {
    const { orderId, amount } = req.body;
    const result = await processPayment(orderId, amount);

    res.json({
      success: true,
      correlationId: req.correlationId,
      data: result
    });
  } catch (error) {
    res.status(500).json({
      success: false,
      correlationId: req.correlationId,
      error: error.message
    });
  }
});

app.listen(3000, () => logger.info('Payment service started'));

2. Python/FastAPI Implementation

# middleware/correlation_id.py
import uuid
from contextvars import ContextVar
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

correlation_id_var: ContextVar[str] = ContextVar('correlation_id', default=None)

def get_correlation_id() -> str:
    return correlation_id_var.get()

class CorrelationIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract or generate correlation ID
        correlation_id = request.headers.get(
            'x-correlation-id',
            str(uuid.uuid4())
        )

        # Set in context
        token = correlation_id_var.set(correlation_id)

        # Process request
        response = await call_next(request)

        # Add to response headers
        response.headers['x-correlation-id'] = correlation_id

        # Reset context
        correlation_id_var.reset(token)

        return response

# logger/logger.py
import logging
import json
from middleware.correlation_id import get_correlation_id

class CorrelationIdFilter(logging.Filter):
    def filter(self, record):
        record.correlation_id = get_correlation_id()
        return True

def setup_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    handler = logging.StreamHandler()

    formatter = logging.Formatter(
        '[%(asctime)s] [%(correlation_id)s] [%(levelname)s]: %(message)s'
    )

    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.addFilter(CorrelationIdFilter())

    return logger

# services/payment_service.py
import httpx
from logger.logger import setup_logger
from middleware.correlation_id import get_correlation_id

logger = setup_logger(__name__)

async def process_payment(order_id: str, amount: float) -> dict:
    correlation_id = get_correlation_id()

    logger.info(f"Processing payment for order {order_id}")

    async with httpx.AsyncClient() as client:
        try:
            # Call inventory service with correlation ID
            response = await client.post(
                'http://inventory-service/api/reserve',
                json={'order_id': order_id, 'quantity': 1},
                headers={'x-correlation-id': correlation_id}
            )
            response.raise_for_status()

            logger.info("Inventory reserved successfully")
            return response.json()

        except httpx.HTTPError as e:
            logger.error(f"Payment processing failed: {str(e)}")
            raise

# main.py
from fastapi import FastAPI, Request
from middleware.correlation_id import CorrelationIdMiddleware
from services.payment_service import process_payment
from logger.logger import setup_logger

app = FastAPI()
app.add_middleware(CorrelationIdMiddleware)

logger = setup_logger(__name__)

@app.post('/api/checkout')
async def checkout(request: Request, order_id: str, amount: float):
    try:
        result = await process_payment(order_id, amount)
        return {
            'success': True,
            'correlation_id': request.headers.get('x-correlation-id'),
            'data': result
        }
    except Exception as e:
        return {
            'success': False,
            'correlation_id': request.headers.get('x-correlation-id'),
            'error': str(e)
        }

3. Java/Spring Boot Implementation

// config/CorrelationIdFilter.java
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.UUID;

@Component
public class CorrelationIdFilter extends OncePerRequestFilter {

    public static final String CORRELATION_ID_HEADER = "x-correlation-id";
    public static final String CORRELATION_ID_LOG_VAR = "correlationId";

    @Override
    protected void doFilterInternal(
        HttpServletRequest request,
        HttpServletResponse response,
        FilterChain filterChain
    ) throws ServletException, IOException {

        String correlationId = request.getHeader(CORRELATION_ID_HEADER);
        if (correlationId == null || correlationId.isEmpty()) {
            correlationId = UUID.randomUUID().toString();
        }

        // Store in MDC (Mapped Diagnostic Context)
        org.slf4j.MDC.put(CORRELATION_ID_LOG_VAR, correlationId);

        response.setHeader(CORRELATION_ID_HEADER, correlationId);

        try {
            filterChain.doFilter(request, response);
        } finally {
            org.slf4j.MDC.remove(CORRELATION_ID_LOG_VAR);
        }
    }
}

// service/PaymentService.java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.client.RestTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpEntity;

@Service
public class PaymentService {

    private static final Logger logger = LoggerFactory.getLogger(PaymentService.class);

    @Autowired
    private RestTemplate restTemplate;

    public void processPayment(String orderId, double amount) {
        String correlationId = org.slf4j.MDC.get("correlationId");

        logger.info("Processing payment for order: {}", orderId);

        try {
            // Create headers with correlation ID
            HttpHeaders headers = new HttpHeaders();
            headers.set("x-correlation-id", correlationId);

            HttpEntity<String> entity = new HttpEntity<>(headers);

            // Call inventory service
            restTemplate.postForObject(
                "http://inventory-service/api/reserve",
                entity,
                String.class
            );

            logger.info("Inventory reserved successfully");

        } catch (Exception e) {
            logger.error("Payment processing failed", e);
            throw new RuntimeException(e);
        }
    }
}

<!-- logback-spring.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>
                [%d{yyyy-MM-dd HH:mm:ss}] [%X{correlationId}] [%level] %logger{36} - %msg%n
            </pattern>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>
</configuration>

4. Go/Gin Implementation

// middleware/correlation_id.go
package middleware

import (
    "context"
    "github.com/gin-gonic/gin"
    "github.com/google/uuid"
)

const CorrelationIDKey = "x-correlation-id"

func CorrelationIdMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        correlationId := c.GetHeader(CorrelationIDKey)
        if correlationId == "" {
            correlationId = uuid.New().String()
        }

        // Store in context
        c.Set(CorrelationIDKey, correlationId)
        c.Request = c.Request.WithContext(
            context.WithValue(c.Request.Context(), CorrelationIDKey, correlationId),
        )

        // Add to response header
        c.Header(CorrelationIDKey, correlationId)

        c.Next()
    }
}

func GetCorrelationId(c *gin.Context) string {
    if id, exists := c.Get(CorrelationIDKey); exists {
        return id.(string)
    }
    return ""
}

// logger/logger.go
package logger

import (
    "context"
    "fmt"
    "log"
)

type ContextLogger struct{}

func (cl *ContextLogger) Info(ctx context.Context, msg string, args ...interface{}) {
    correlationId := ctx.Value("x-correlation-id")
    log.Printf("[%s] INFO: %s\n", correlationId, fmt.Sprintf(msg, args...))
}

func (cl *ContextLogger) Error(ctx context.Context, msg string, args ...interface{}) {
    correlationId := ctx.Value("x-correlation-id")
    log.Printf("[%s] ERROR: %s\n", correlationId, fmt.Sprintf(msg, args...))
}

// service/payment_service.go
package service

import (
    "bytes"
    "context"
    "encoding/json"
    "net/http"
    "payment-service/logger"
)

func ProcessPayment(ctx context.Context, orderId string, amount float64) error {
    correlationId := ctx.Value("x-correlation-id").(string)

    logger := &logger.ContextLogger{}
    logger.Info(ctx, "Processing payment for order: %s", orderId)

    // Prepare request to inventory service
    payload := map[string]interface{}{
        "order_id": orderId,
        "quantity": 1,
    }

    body, _ := json.Marshal(payload)
    req, _ := http.NewRequestWithContext(
        ctx,
        "POST",
        "http://inventory-service/api/reserve",
        bytes.NewBuffer(body),
    )

    // Add correlation ID header
    req.Header.Set("x-correlation-id", correlationId)
    req.Header.Set("Content-Type", "application/json")

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        logger.Error(ctx, "Payment processing failed: %v", err)
        return err
    }
    defer resp.Body.Close()

    logger.Info(ctx, "Inventory reserved successfully")
    return nil
}

// main.go
package main

import (
    "github.com/gin-gonic/gin"
    "payment-service/middleware"
    "payment-service/service"
)

func main() {
    router := gin.Default()

    // Apply correlation ID middleware
    router.Use(middleware.CorrelationIdMiddleware())

    router.POST("/api/checkout", func(c *gin.Context) {
        var req struct {
            OrderId string  `json:"order_id"`
            Amount  float64 `json:"amount"`
        }

        c.BindJSON(&req)

        err := service.ProcessPayment(c.Request.Context(), req.OrderId, req.Amount)

        if err != nil {
            c.JSON(500, gin.H{
                "success":        false,
                "correlation_id": middleware.GetCorrelationId(c),
                "error":          err.Error(),
            })
            return
        }

        c.JSON(200, gin.H{
            "success":        true,
            "correlation_id": middleware.GetCorrelationId(c),
        })
    })

    router.Run(":3000")
}

---

Advanced Tips & Best Practices

1. Distributed Tracing Integration

// Integrate with Jaeger/Zipkin
const initTracer = require('jaeger-client').initTracer;

const tracer = initTracer({
  serviceName: 'payment-service',
  sampler: { type: 'const', param: 1 },
  reporter: { logSpans: true }
});

app.use((req, res, next) => {
  const wireCtx = tracer.extract('http_headers', req.headers);
  const span = tracer.startSpan('http_request', { childOf: wireCtx });

  req.span = span;
  req.correlationId = span.getTraceID().toString();

  next();
});

2. Correlation ID Propagation in Message Queues

// RabbitMQ
async function publishEvent(event, data) {
  const correlationId = getCorrelationId();

  await channel.assertQueue('events');
  channel.sendToQueue('events', Buffer.from(JSON.stringify(data)), {
    correlationId: correlationId,
    headers: { 'x-correlation-id': correlationId }
  });
}

// Consumer
channel.consume('events', (msg) => {
  const correlationId = msg.properties.correlationId;
  asyncLocalStorage.run(correlationId, () => {
    processEvent(JSON.parse(msg.content.toString()));
  });
});

3. Correlation ID in Database Queries

// Store correlation ID with database records
async function saveOrder(orderId, data) {
  const correlationId = getCorrelationId();

  await db.query(
    'INSERT INTO orders (id, data, correlation_id, created_at) VALUES (?, ?, ?, ?)',
    [orderId, JSON.stringify(data), correlationId, new Date()]
  );
}

// Query by correlation ID for debugging
async function getOrdersByCorrelationId(correlationId) {
  return db.query(
    'SELECT * FROM orders WHERE correlation_id = ?',
    [correlationId]
  );
}

4. Correlation ID in Error Tracking

// Sentry integration
Sentry.init({ dsn: 'your-dsn' });

app.use((err, req, res, next) => {
  Sentry.captureException(err, {
    tags: {
      correlationId: req.correlationId
    },
    extra: {
      correlationId: req.correlationId,
      userId: req.user?.id
    }
  });

  res.status(500).json({
    error: 'Internal server error',
    correlationId: req.correlationId,
    sentryId: Sentry.lastEventId()
  });
});

5. Correlation ID Retention Policy

```javascript // Clean up old correlation IDs from cache const redis = require('redis'); const client = redis.createClient();

async function storeCorrelationId(correlationId, metadata) { // Store with 24-