ELK Stack — Project: Multi-Source Log Aggregation Pipeline

Project Overview

Real production environments rarely have a single log source. In this project you will build a multi-pipeline Logstash setup that processes logs from three completely different sources:

• Apache HTTP Server — classic combined log format
• Syslog — system-level messages via UDP/TCP
• JSON Application Logs — custom application emitting structured logs

Each source has its own Logstash pipeline with dedicated parsing, enrichment, and output configuration. You will also implement a dead letter queue, monitoring, and conditional routing.

---

Architecture

┌──────────────────┐
│  Apache HTTPD    │──access.log──▶ Filebeat ──▶ Logstash (Pipeline 1) ──┐
│                  │──error.log──▶                                       │
└──────────────────┘                                                     │
                                                                         ▼
┌──────────────────┐                                             ┌───────────────┐
│  Linux Syslog    │──UDP:5514──▶ Logstash (Pipeline 2) ────────▶│ Elasticsearch │
│  (rsyslog)       │                                             │               │
└──────────────────┘                                             │  Indices:     │
                                                                 │  apache-*     │
┌──────────────────┐                                             │  syslog-*     │
│  JSON App        │──TCP:5045──▶ Logstash (Pipeline 3) ────────▶│  app-*        │
│  (custom app)    │                                             └───────┬───────┘
└──────────────────┘                                                     │
                                                                         ▼
                                                                 ┌─────────────┐
                                                                 │   Kibana     │
                                                                 └─────────────┘

---

Project Structure

multi-source-elk/
├── docker-compose.yml
├── apache/
│   └── httpd.conf
├── json-app/
│   ├── app.py
│   ├── requirements.txt
│   └── Dockerfile
├── filebeat/
│   └── filebeat.yml
├── logstash/
│   ├── pipelines.yml
│   ├── pipeline-apache/
│   │   └── apache.conf
│   ├── pipeline-syslog/
│   │   └── syslog.conf
│   ├── pipeline-json/
│   │   └── json-app.conf
│   └── patterns/
│       └── custom-patterns
└── syslog-generator/
    └── generate.sh

---

Step 1 — 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/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./logstash/pipeline-apache:/usr/share/logstash/pipeline-apache:ro
      - ./logstash/pipeline-syslog:/usr/share/logstash/pipeline-syslog:ro
      - ./logstash/pipeline-json:/usr/share/logstash/pipeline-json:ro
      - ./logstash/patterns:/usr/share/logstash/patterns:ro
      - logstash-dlq:/usr/share/logstash/data/dead_letter_queue
    ports:
      - "5044:5044"
      - "5045:5045"
      - "5514:5514/udp"
      - "9600:9600"
    environment:
      - "LS_JAVA_OPTS=-Xms512m -Xmx512m"
    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

  apache:
    image: httpd:2.4-alpine
    container_name: apache-web
    volumes:
      - ./apache/httpd.conf:/usr/local/apache2/conf/httpd.conf:ro
      - apache-logs:/usr/local/apache2/logs
    ports:
      - "8080:80"
    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
      - apache-logs:/var/log/apache2:ro
    networks:
      - elk
    depends_on:
      - logstash
      - apache

  json-app:
    build: ./json-app
    container_name: json-app
    networks:
      - elk
    depends_on:
      - logstash

  syslog-generator:
    image: alpine:3.19
    container_name: syslog-generator
    volumes:
      - ./syslog-generator/generate.sh:/generate.sh:ro
    entrypoint: ["/bin/sh", "/generate.sh"]
    networks:
      - elk
    depends_on:
      - logstash

volumes:
  es-data:
  apache-logs:
  logstash-dlq:

networks:
  elk:
    driver: bridge

---

Step 2 — Logstash Multi-Pipeline Configuration

Create logstash/pipelines.yml:

- pipeline.id: apache
  path.config: "/usr/share/logstash/pipeline-apache/apache.conf"
  pipeline.workers: 2
  dead_letter_queue.enable: true

- pipeline.id: syslog
  path.config: "/usr/share/logstash/pipeline-syslog/syslog.conf"
  pipeline.workers: 1

- pipeline.id: json-app
  path.config: "/usr/share/logstash/pipeline-json/json-app.conf"
  pipeline.workers: 1
  dead_letter_queue.enable: true

---

Step 3 — Custom Grok Patterns

Create logstash/patterns/custom-patterns:

# Custom Apache error log pattern
APACHE_ERROR_LOG \[%{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}\] \[%{WORD:module}:%{LOGLEVEL:log_level}\] \[pid %{NUMBER:pid}\] %{GREEDYDATA:error_message}

# Custom application log pattern
APP_LOG \[%{TIMESTAMP_ISO8601:timestamp}\] \[%{LOGLEVEL:level}\] \[%{DATA:module}\] %{GREEDYDATA:message}

---

Step 4 — Apache Pipeline

Create logstash/pipeline-apache/apache.conf:

input {
  beats {
    port => 5044
    type => "apache"
  }
}

filter {
  if [fields][log_type] == "apache_access" {
    grok {
      match => {
        "message" => '%{COMBINEDAPACHELOG}'
      }
    }

    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }

    mutate {
      convert => {
        "response" => "integer"
        "bytes"    => "integer"
      }
    }

    geoip {
      source => "clientip"
      target => "geoip"
    }

    useragent {
      source => "agent"
      target => "ua"
    }

    # Categorize response codes
    if [response] >= 500 {
      mutate { add_field => { "response_category" => "server_error" } }
    } else if [response] >= 400 {
      mutate { add_field => { "response_category" => "client_error" } }
    } else if [response] >= 300 {
      mutate { add_field => { "response_category" => "redirect" } }
    } else {
      mutate { add_field => { "response_category" => "success" } }
    }

    mutate {
      add_field => { "[@metadata][index]" => "apache-access" }
      remove_field => ["message"]
    }
  }

  else if [fields][log_type] == "apache_error" {
    grok {
      patterns_dir => ["/usr/share/logstash/patterns"]
      match => {
        "message" => "%{APACHE_ERROR_LOG}"
      }
    }

    mutate {
      add_field => { "[@metadata][index]" => "apache-error" }
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://es01:9200"]
    index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
  }
}

---

Step 5 — Syslog Pipeline

Create logstash/pipeline-syslog/syslog.conf:

input {
  udp {
    port => 5514
    type => "syslog"
  }
}

filter {
  grok {
    match => {
      "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_host} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}"
    }
  }

  date {
    match => ["syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
    target => "@timestamp"
    remove_field => ["syslog_timestamp"]
  }

  # Enrich with severity level
  if [syslog_message] =~ /(?i)error|fail|critical/ {
    mutate { add_field => { "severity" => "high" } }
  } else if [syslog_message] =~ /(?i)warn/ {
    mutate { add_field => { "severity" => "medium" } }
  } else {
    mutate { add_field => { "severity" => "low" } }
  }

  # Tag SSH-related messages
  if [syslog_program] == "sshd" {
    mutate { add_tag => ["ssh"] }

    if [syslog_message] =~ /Failed password/ {
      grok {
        match => {
          "syslog_message" => "Failed password for (invalid user )?%{USERNAME:ssh_user} from %{IP:ssh_source_ip}"
        }
      }
      mutate { add_tag => ["failed_login"] }
    }
  }

  mutate {
    remove_field => ["message"]
  }
}

output {
  elasticsearch {
    hosts => ["http://es01:9200"]
    index => "syslog-%{+YYYY.MM.dd}"
  }
}

---

Step 6 — JSON Application Pipeline

Create logstash/pipeline-json/json-app.conf:

input {
  tcp {
    port => 5045
    codec => json_lines
    type => "json-app"
  }
}

filter {
  # The input is already JSON, so fields are parsed
  if [timestamp] {
    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }
  }

  # Conditional routing based on log level
  if [level] == "ERROR" or [level] == "CRITICAL" {
    mutate {
      add_tag => ["alert"]
      add_field => { "priority" => "high" }
    }
  }

  # Enrich with environment info
  mutate {
    add_field => {
      "pipeline" => "json-app"
      "environment" => "development"
    }
  }
}

output {
  if "alert" in [tags] {
    elasticsearch {
      hosts => ["http://es01:9200"]
      index => "app-alerts-%{+YYYY.MM.dd}"
    }
  }

  elasticsearch {
    hosts => ["http://es01:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
  }
}

---

Step 7 — Filebeat for Apache Logs

Create filebeat/filebeat.yml:

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/apache2/access_log
    fields:
      log_type: apache_access
    fields_under_root: false

  - type: log
    enabled: true
    paths:
      - /var/log/apache2/error_log
    fields:
      log_type: apache_error
    fields_under_root: false

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

logging.level: info

---

Step 8 — JSON Application

Create json-app/app.py:

import json
import socket
import time
import random
import datetime

LOGSTASH_HOST = "logstash"
LOGSTASH_PORT = 5045

MODULES = ["auth", "payment", "inventory", "notification", "scheduler"]
LEVELS = ["DEBUG", "INFO", "INFO", "INFO", "WARN", "ERROR", "CRITICAL"]
MESSAGES = {
    "auth": ["User login successful", "User login failed", "Token refreshed", "Session expired"],
    "payment": ["Payment processed", "Payment failed", "Refund initiated", "Invoice generated"],
    "inventory": ["Stock updated", "Low stock alert", "Item added", "Item removed"],
    "notification": ["Email sent", "SMS sent", "Push notification sent", "Notification failed"],
    "scheduler": ["Job started", "Job completed", "Job failed", "Job scheduled"],
}

def send_log(sock, log_entry):
    try:
        data = json.dumps(log_entry) + "\n"
        sock.sendall(data.encode("utf-8"))
    except (BrokenPipeError, ConnectionResetError):
        return False
    return True

def main():
    while True:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((LOGSTASH_HOST, LOGSTASH_PORT))
            print(f"Connected to Logstash at {LOGSTASH_HOST}:{LOGSTASH_PORT}")

            while True:
                module = random.choice(MODULES)
                level = random.choice(LEVELS)
                message = random.choice(MESSAGES[module])

                log_entry = {
                    "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
                    "level": level,
                    "module": module,
                    "message": message,
                    "request_id": f"req-{random.randint(10000, 99999)}",
                    "user_id": f"user-{random.randint(1, 100)}",
                    "duration_ms": random.randint(1, 5000),
                }

                if not send_log(sock, log_entry):
                    break

                time.sleep(random.uniform(0.5, 3.0))

        except ConnectionRefusedError:
            print("Logstash not ready, retrying in 5s...")
            time.sleep(5)
        except Exception as e:
            print(f"Error: {e}, retrying in 5s...")
            time.sleep(5)
        finally:
            sock.close()

if __name__ == "__main__":
    main()

Create json-app/requirements.txt:

# No external dependencies needed — uses stdlib only

Create json-app/Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY app.py .
CMD ["python", "-u", "app.py"]

---

Step 9 — Syslog Generator

Create syslog-generator/generate.sh:

#!/bin/sh

echo "Starting syslog generator..."
sleep 10  # Wait for Logstash to be ready

FACILITIES=("kern" "user" "mail" "daemon" "auth" "syslog" "cron")
PROGRAMS=("sshd" "sudo" "cron" "systemd" "kernel" "NetworkManager")

while true; do
  PROGRAM=${PROGRAMS[$((RANDOM % ${#PROGRAMS[@]}))]}
  MONTH=$(date +%b)
  DAY=$(date +%d)
  TIME=$(date +%H:%M:%S)
  HOST="server-$(( RANDOM % 5 + 1 ))"

  case $PROGRAM in
    sshd)
      MSGS=(
        "Accepted publickey for admin from 10.0.1.$((RANDOM % 255)) port $((RANDOM % 50000 + 1024))"
        "Failed password for invalid user test from 192.168.1.$((RANDOM % 255)) port $((RANDOM % 50000 + 1024))"
        "Connection closed by 10.0.2.$((RANDOM % 255)) port $((RANDOM % 50000 + 1024))"
      )
      ;;
    sudo)
      MSGS=(
        "admin : TTY=pts/0 ; PWD=/home/admin ; USER=root ; COMMAND=/bin/systemctl restart nginx"
        "deploy : TTY=pts/1 ; PWD=/opt/app ; USER=root ; COMMAND=/bin/docker-compose up -d"
      )
      ;;
    *)
      MSGS=(
        "Service started successfully"
        "Configuration reloaded"
        "Warning: high memory usage detected"
        "Error: connection timeout to database"
      )
      ;;
  esac

  MSG=${MSGS[$((RANDOM % ${#MSGS[@]}))]}
  SYSLOG_MSG="$MONTH $DAY $TIME $HOST $PROGRAM[$$]: $MSG"

  echo "$SYSLOG_MSG" | nc -u -w1 logstash 5514

  sleep $((RANDOM % 5 + 1))
done

---

Step 10 — Launch and Verify

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

# Wait for services
sleep 30

# Generate Apache traffic
for i in $(seq 1 100); do
  curl -s http://localhost:8080/ > /dev/null
  curl -s http://localhost:8080/nonexistent > /dev/null 2>&1
done

# Check Logstash pipeline stats
curl -s http://localhost:9600/_node/stats/pipelines?pretty | head -80

# Check Elasticsearch indices
curl -s http://localhost:9200/_cat/indices?v | grep -E "apache|syslog|app"

---

Step 11 — Monitoring the Dead Letter Queue

# Check DLQ stats via Logstash API
curl -s http://localhost:9600/_node/stats/pipelines?pretty | grep -A5 dead_letter_queue

---

Step 12 — Kibana Dashboards

Data Views

Create data views for:

• apache-access-*
• syslog-*
• app-logs-*
• app-alerts-*

Dashboard Panels

Panel	Source	Type	Description
All Sources Log Volume	All indices	Stacked area	Compare volume across sources
Apache Response Codes	apache-access-*	Pie chart	Terms on response
Syslog Severity	syslog-*	Gauge	Count by severity
App Error Rate	app-logs-*	Line chart	Filter level: ERROR over time
Failed SSH Logins	syslog-*	Metric	Filter tag failed_login
High Priority Alerts	app-alerts-*	Data table	All alert documents
Top App Modules	app-logs-*	Horizontal bar	Terms on module

---

Cleanup

docker-compose down -v --rmi local

---

What You Learned

• How to configure multiple Logstash pipelines with pipelines.yml
• How to write custom grok patterns
• How to use conditional routing in Logstash outputs
• How to enable and monitor the dead letter queue
• How to aggregate logs from completely different source types

---

Next Project

→ Project 5 — Security Event Dashboard (SIEM-lite)