ELK Stack — Project: Microservices Centralized Logging

Project Overview

Microservices architectures produce logs from many independent services. Without centralized logging, tracing a single user request across services becomes nearly impossible.

In this project you will build:

• Three sample microservices that emit structured JSON logs
• Correlation IDs that propagate across service-to-service calls
• Filebeat with the container log input to collect Docker logs
• Logstash to process and route JSON logs
• Kibana dashboards to trace requests across services

---

Architecture

┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│  API Gateway │─▶│ Order Service│─▶│Payment Service│
│  (Python)    │  │  (Python)    │  │  (Node.js)   │
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       │ JSON logs       │ JSON logs       │ JSON logs
       ▼                 ▼                 ▼
   ┌─────────────────────────────────────────┐
   │           Docker Log Driver             │
   └─────────────────┬───────────────────────┘
                     ▼
              ┌───────────┐
              │ Filebeat   │
              │ (container)│
              └─────┬─────┘
                    ▼
              ┌───────────┐     ┌───────────────┐     ┌─────────┐
              │ Logstash   │────▶│ Elasticsearch │────▶│ Kibana  │
              └───────────┘     └───────────────┘     └─────────┘

---

Project Structure

microservices-elk/
├── docker-compose.yml
├── api-gateway/
│   ├── app.py
│   ├── requirements.txt
│   └── Dockerfile
├── order-service/
│   ├── app.py
│   ├── requirements.txt
│   └── Dockerfile
├── payment-service/
│   ├── app.js
│   ├── package.json
│   └── Dockerfile
├── filebeat/
│   └── filebeat.yml
└── logstash/
    └── pipeline/
        └── microservices.conf

---

Step 1 — API Gateway (Python)

Create api-gateway/app.py:

import json
import logging
import uuid
import time
from flask import Flask, request, jsonify, g
import requests

app = Flask(__name__)

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "service": "api-gateway",
            "message": record.getMessage(),
            "correlation_id": getattr(record, "correlation_id", "unknown"),
            "logger": record.name,
        }
        return json.dumps(log_entry)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger("api-gateway")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

@app.before_request
def before_request():
    g.correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
    g.start_time = time.time()

@app.after_request
def after_request(response):
    duration = time.time() - g.start_time
    logger.info(
        f"Request completed: {request.method} {request.path} "
        f"status={response.status_code} duration={duration:.3f}s",
        extra={"correlation_id": g.correlation_id},
    )
    return response

@app.route("/api/order", methods=["POST"])
def create_order():
    correlation_id = g.correlation_id
    logger.info("Received order request", extra={"correlation_id": correlation_id})

    # Call order service
    try:
        resp = requests.post(
            "http://order-service:5001/orders",
            json=request.get_json(),
            headers={"X-Correlation-ID": correlation_id},
            timeout=5,
        )
        return jsonify(resp.json()), resp.status_code
    except requests.exceptions.RequestException as e:
        logger.error(f"Order service call failed: {e}",
                     extra={"correlation_id": correlation_id})
        return jsonify({"error": "Service unavailable"}), 503

@app.route("/health")
def health():
    return jsonify({"status": "healthy", "service": "api-gateway"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Create api-gateway/requirements.txt:

flask==3.0.0
requests==2.31.0

Create api-gateway/Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
CMD ["python", "app.py"]

---

Step 2 — Order Service (Python)

Create order-service/app.py:

import json
import logging
import time
from flask import Flask, request, jsonify, g
import requests

app = Flask(__name__)

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "service": "order-service",
            "message": record.getMessage(),
            "correlation_id": getattr(record, "correlation_id", "unknown"),
        }
        return json.dumps(log_entry)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger("order-service")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

@app.before_request
def before_request():
    g.correlation_id = request.headers.get("X-Correlation-ID", "unknown")
    g.start_time = time.time()

@app.route("/orders", methods=["POST"])
def create_order():
    cid = g.correlation_id
    data = request.get_json()
    logger.info(f"Processing order: {data}", extra={"correlation_id": cid})

    # Simulate order processing
    time.sleep(0.1)
    order_id = "ORD-" + str(int(time.time()))

    # Call payment service
    try:
        resp = requests.post(
            "http://payment-service:5002/payments",
            json={"order_id": order_id, "amount": data.get("amount", 0)},
            headers={"X-Correlation-ID": cid},
            timeout=5,
        )
        payment_status = resp.json().get("status", "unknown")
    except requests.exceptions.RequestException as e:
        logger.error(f"Payment service call failed: {e}",
                     extra={"correlation_id": cid})
        payment_status = "failed"

    logger.info(f"Order {order_id} created, payment={payment_status}",
                extra={"correlation_id": cid})

    return jsonify({
        "order_id": order_id,
        "payment_status": payment_status,
        "correlation_id": cid,
    }), 201

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5001)

Create order-service/requirements.txt:

flask==3.0.0
requests==2.31.0

Create order-service/Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
CMD ["python", "app.py"]

---

Step 3 — Payment Service (Node.js)

Create payment-service/app.js:

const express = require("express");
const app = express();
app.use(express.json());

function log(level, message, correlationId, extra = {}) {
  const entry = {
    timestamp: new Date().toISOString(),
    level,
    service: "payment-service",
    message,
    correlation_id: correlationId || "unknown",
    ...extra,
  };
  console.log(JSON.stringify(entry));
}

app.post("/payments", (req, res) => {
  const correlationId = req.headers["x-correlation-id"] || "unknown";
  const { order_id, amount } = req.body;

  log("INFO", `Processing payment for order ${order_id}, amount=${amount}`, correlationId);

  // Simulate payment processing
  const success = Math.random() > 0.1;
  const status = success ? "completed" : "failed";

  if (!success) {
    log("ERROR", `Payment failed for order ${order_id}`, correlationId, { amount });
  } else {
    log("INFO", `Payment completed for order ${order_id}`, correlationId, { amount });
  }

  res.json({
    order_id,
    status,
    transaction_id: `TXN-${Date.now()}`,
    correlation_id: correlationId,
  });
});

app.get("/health", (req, res) => {
  res.json({ status: "healthy", service: "payment-service" });
});

app.listen(5002, () => {
  log("INFO", "Payment service started on port 5002", "system");
});

Create payment-service/package.json:

{
  "name": "payment-service",
  "version": "1.0.0",
  "main": "app.js",
  "dependencies": {
    "express": "^4.18.2"
  }
}

Create payment-service/Dockerfile:

FROM node:20-alpine
WORKDIR /app
COPY package.json .
RUN npm install --production
COPY app.js .
CMD ["node", "app.js"]

---

Step 4 — Filebeat Configuration

Create filebeat/filebeat.yml:

filebeat.inputs:
  - type: container
    paths:
      - /var/lib/docker/containers/*/*.log
    processors:
      - add_docker_metadata:
          host: "unix:///var/run/docker.sock"

output.logstash:
  hosts: ["logstash:5044"]

logging.level: info

---

Step 5 — Logstash Pipeline

Create logstash/pipeline/microservices.conf:

input {
  beats {
    port => 5044
  }
}

filter {
  # Parse the JSON log message from the container
  if [message] {
    json {
      source => "message"
      target => "app"
      skip_on_invalid_json => true
    }
  }

  # Only process logs from our microservices
  if [container][name] in ["api-gateway", "order-service", "payment-service"] {
    # Promote fields from the parsed JSON
    if [app][service] {
      mutate {
        add_field => {
          "service_name"   => "%{[app][service]}"
          "correlation_id" => "%{[app][correlation_id]}"
          "log_level"      => "%{[app][level]}"
          "log_message"    => "%{[app][message]}"
        }
        remove_field => ["message"]
      }
    }
  }
}

output {
  if [service_name] {
    elasticsearch {
      hosts => ["http://es01:9200"]
      index => "microservices-%{[service_name]}-%{+YYYY.MM.dd}"
    }
  } else {
    elasticsearch {
      hosts => ["http://es01:9200"]
      index => "other-logs-%{+YYYY.MM.dd}"
    }
  }
}

---

Step 6 — Docker Compose

Create docker-compose.yml:

version: "3.8"

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0
    container_name: es01
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ports:
      - "9200:9200"
    volumes:
      - es-data:/usr/share/elasticsearch/data
    networks:
      - elk
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:9200/_cluster/health || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 10

  logstash:
    image: docker.elastic.co/logstash/logstash:8.13.0
    container_name: logstash
    volumes:
      - ./logstash/pipeline/microservices.conf:/usr/share/logstash/pipeline/microservices.conf:ro
    ports:
      - "5044:5044"
    environment:
      - "LS_JAVA_OPTS=-Xms256m -Xmx256m"
    networks:
      - elk
    depends_on:
      elasticsearch:
        condition: service_healthy

  kibana:
    image: docker.elastic.co/kibana/kibana:8.13.0
    container_name: kibana
    environment:
      - ELASTICSEARCH_HOSTS=http://es01:9200
    ports:
      - "5601:5601"
    networks:
      - elk
    depends_on:
      elasticsearch:
        condition: service_healthy

  api-gateway:
    build: ./api-gateway
    container_name: api-gateway
    ports:
      - "5000:5000"
    networks:
      - elk
    depends_on:
      - order-service

  order-service:
    build: ./order-service
    container_name: order-service
    ports:
      - "5001:5001"
    networks:
      - elk
    depends_on:
      - payment-service

  payment-service:
    build: ./payment-service
    container_name: payment-service
    ports:
      - "5002:5002"
    networks:
      - elk

  filebeat:
    image: docker.elastic.co/beats/filebeat:8.13.0
    container_name: filebeat
    user: root
    volumes:
      - ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - /var/run/docker.sock:/var/run/docker.sock:ro
    networks:
      - elk
    depends_on:
      - logstash

volumes:
  es-data:

networks:
  elk:
    driver: bridge

---

Step 7 — Launch and Test

# Build and start everything
docker-compose up -d --build

# Wait for services to be healthy
sleep 30

# Generate test traffic
for i in $(seq 1 50); do
  curl -s -X POST http://localhost:5000/api/order \
    -H "Content-Type: application/json" \
    -d "{\"item\": \"widget-$i\", \"amount\": $((RANDOM % 100 + 1))}" > /dev/null
  sleep 0.5
done

echo "Generated 50 orders"

---

Step 8 — Verify Data

Check indices

curl -s http://localhost:9200/_cat/indices?v | grep microservices

Search by correlation ID

# Get a correlation ID from recent logs
CID=$(curl -s http://localhost:9200/microservices-*/_search -H 'Content-Type: application/json' -d'
{
  "size": 1,
  "_source": ["correlation_id"]
}' | python3 -c "import sys,json; print(json.load(sys.stdin)['hits']['hits'][0]['_source']['correlation_id'])")

echo "Tracing correlation ID: $CID"

# Find all logs for this correlation ID
curl -s "http://localhost:9200/microservices-*/_search?pretty" -H 'Content-Type: application/json' -d"
{
  \"query\": {
    \"match\": { \"correlation_id\": \"$CID\" }
  },
  \"sort\": [{ \"@timestamp\": \"asc\" }]
}"

---

Step 9 — Kibana Dashboards

Data Views

Create data views for:

• microservices-api-gateway-*
• microservices-order-service-*
• microservices-payment-service-*
• microservices-* (all services combined)

Dashboard Panels

Panel	Type	Purpose
Log Volume by Service	Stacked bar	Compare log volume across services
Error Rate by Service	Line chart	Filter log_level: ERROR
Request Flow (Correlation)	Data table	Group by correlation_id, show services
Payment Success Rate	Gauge	Ratio of successful vs failed payments
Latency Over Time	Line chart	Based on duration field if available
Top Error Messages	Tag cloud	Terms on log_message where log_level: ERROR

---

Troubleshooting

Symptom	Cause	Fix
Empty indices	Filebeat can’t access Docker socket	Check Filebeat volume mounts and permissions
JSON parse errors	Container logs include non-JSON output	Add skip_on_invalid_json in Logstash
Missing correlation IDs	Header not propagated	Verify X-Correlation-ID header in service code

---

Cleanup

docker-compose down -v --rmi local

---

What You Learned

• How to implement structured JSON logging across multiple languages
• How to propagate correlation IDs for distributed tracing
• How to use Filebeat container input for Docker log collection
• How to build cross-service dashboards in Kibana

---

Next Project

→ Project 3 — ELK on Kubernetes with ECK