Build a Log Aggregation and Anomaly Detection Pipeline with ELK and Kafka

Deploy ELK stack on Kubernetes, ship logs via Filebeat DaemonSet, buffer with Kafka for traffic spikes, and alert on error rate anomalies with Watcher.

Related Concepts & TermsProcess grep ConfigMap DaemonSet Helm Daemon

Domains & Technologies

Domains

ELASTICSEARCHKAFKAMONITORING

Technologies

KUBERNETES

Blueprint Walkthrough

Architecture Overview

This project builds a production-grade log aggregation pipeline. Every pod in your Kubernetes cluster generates logs. Without a centralised system, those logs live only in the pod — when the pod is deleted or crashes, the logs are gone forever. That is the worst possible time to lose logs — exactly when you need them most for debugging.

The ELK stack (Elasticsearch for storage and search, Logstash for processing, Kibana for visualisation) solves this. Kafka sits between log producers and Elasticsearch as a buffer, preventing Elasticsearch from being overwhelmed during traffic spikes — the exact scenario Swiggy faces at 1pm every weekday when order volume spikes 10x.

JAVA

    Kubernetes Pods
    (generate logs)
          |
    Filebeat DaemonSet
    (one per node, collects all logs)
          |
          v
    Apache Kafka
    (buffer — absorbs spikes)
          |
          v
    Logstash
    (parse, enrich, filter logs)
          |
          v
    Elasticsearch
    (store and index)
          |
          v
    Kibana
    (dashboards and alerts)

Problem Solved

Without a log pipeline, production debugging means running kubectl logs pod-name and hoping the pod is still running, hoping the logs have not been rotated out, and manually scanning thousands of lines. In a cluster with 50 pods you have no way to search across all of them simultaneously.

With this pipeline, every log line from every pod is available in Kibana within seconds. You can search across all services, all pods, all time — level:ERROR AND service:payment-service AND timestamp:[now-1h TO now] returns every error from the payment service in the last hour across all replicas instantly.

The Kafka buffer solves the thundering herd problem. During Swiggy's 1pm lunch spike, pods emit 100x their normal log volume. Without Kafka, this surge hits Elasticsearch directly and causes write failures, losing logs. Kafka absorbs the burst and feeds Elasticsearch at a steady rate.

Step-by-Step Implementation Guide

Step 1: Deploy Elasticsearch and Kibana

Bash

## Add the Elastic Helm repository
helm repo add elastic https://helm.elastic.co
helm repo update
 
## Create the logging namespace
kubectl create namespace logging
 
## Create elasticsearch-values.yaml
cat > elasticsearch-values.yaml << 'EOF'
replicas: 3  # 3-node cluster for production high availability
minimumMasterNodes: 2  # Quorum — prevents split brain
 
resources:
  requests:
    cpu: 500m
    memory: 2Gi
  limits:
    cpu: 2
    memory: 4Gi
 
volumeClaimTemplate:
  accessModes: ["ReadWriteOnce"]
  resources:
    requests:
      storage: 100Gi  # Each node stores 100GB of logs
  storageClassName: gp2
 
esConfig:
  elasticsearch.yml: |
    cluster.name: "production-logs"
    network.host: 0.0.0.0
    # Index lifecycle management — automatically delete old logs
    xpack.ilm.enabled: true
EOF
 
## Install Elasticsearch
helm install elasticsearch elastic/elasticsearch \
  --namespace logging \
  --values elasticsearch-values.yaml
 
## Wait for all 3 nodes to be ready (takes 3-5 minutes)
kubectl get pods -n logging --watch
## Expected: elasticsearch-master-0, -1, -2 all Running
 
## Install Kibana
helm install kibana elastic/kibana \
  --namespace logging \
  --set elasticsearchHosts="http://elasticsearch-master:9200"
 
## Access Kibana
kubectl port-forward -n logging svc/kibana-kibana 5601:5601
## Open http://localhost:5601

Remember
Elasticsearch is memory-intensive. The JVM heap is set to half the container memory limit by default. With 4Gi memory limit, Elasticsearch gets 2Gi heap. Never set the heap above 31GB — above that the JVM cannot use compressed object pointers and performance degrades.

Step 2: Deploy Apache Kafka as the Log Buffer

Bash

## Add Bitnami repo (has excellent Kafka Helm chart)
helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update
 
## Create kafka-values.yaml
cat > kafka-values.yaml << 'EOF'
replicaCount: 3  # 3 Kafka brokers for high availability
 
kraft:
  enabled: true  # Use KRaft mode — no ZooKeeper dependency
 
persistence:
  enabled: true
  size: 50Gi
  storageClass: gp2
 
resources:
  requests:
    cpu: 500m
    memory: 1Gi
  limits:
    cpu: 2
    memory: 2Gi
 
## Topic configuration for log ingestion
extraEnvVars:
  * name: KAFKA_CFG_LOG_RETENTION_HOURS
    value: "24"  # Keep logs in Kafka for 24 hours
  * name: KAFKA_CFG_LOG_RETENTION_BYTES
    value: "53687091200"  # 50GB max per partition
EOF
 
## Install Kafka
helm install kafka bitnami/kafka \
  --namespace logging \
  --values kafka-values.yaml
 
## Wait for brokers to be ready
kubectl get pods -n logging -l app.kubernetes.io/name=kafka
## Expected: kafka-0, kafka-1, kafka-2 all Running
 
## Create the logs topic
kubectl exec -n logging kafka-0 -- \
  kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --create \
  --topic kubernetes-logs \
  --partitions 6 \
  --replication-factor 3
 
## Verify topic was created
kubectl exec -n logging kafka-0 -- \
  kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --topic kubernetes-logs

Step 3: Deploy Filebeat as a DaemonSet

Filebeat runs one pod per node in your cluster. It monitors all log files on the node and ships them to Kafka.

Bash

## Create filebeat-config.yaml
cat > filebeat-config.yaml << 'EOF'
apiVersion: v1
kind: ConfigMap
metadata:
  name: filebeat-config
  namespace: logging
data:
  filebeat.yml: |
    filebeat.inputs:
    * type: container
      paths:
        * /var/log/containers/*.log  # All pod logs on this node
      processors:
        * add_kubernetes_metadata:
            host: ${NODE_NAME}
            matchers:
            * logs_path:
                logs_path: "/var/log/containers/"
 
    processors:
      * drop_event:
          when:
            contains:
              kubernetes.namespace: "kube-system"  # Skip system logs
      * add_fields:
          target: ''
          fields:
            environment: production
            region: ap-south-1
 
    # Ship to Kafka (not directly to Elasticsearch — always buffer)
    output.kafka:
      hosts: ["kafka.logging.svc.cluster.local:9092"]
      topic: 'kubernetes-logs'
      partition.round_robin:
        reachable_only: false
      required_acks: 1
      compression: gzip
      max_message_bytes: 1000000
EOF
 
kubectl apply -f filebeat-config.yaml
 
## Create the Filebeat DaemonSet
kubectl apply -f - <<EOF
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: filebeat
  namespace: logging
spec:
  selector:
    matchLabels:
      app: filebeat
  template:
    metadata:
      labels:
        app: filebeat
    spec:
      serviceAccountName: filebeat
      terminationGracePeriodSeconds: 30
      containers:
      * name: filebeat
        image: docker.elastic.co/beats/filebeat:8.11.0
        args: ["-c", "/etc/filebeat.yml", "-e"]
        env:
        * name: NODE_NAME
          valueFrom:
            fieldRef:
              fieldPath: spec.nodeName
        resources:
          requests:
            cpu: 100m
            memory: 100Mi
          limits:
            cpu: 1000m
            memory: 500Mi
        volumeMounts:
        * name: config
          mountPath: /etc/filebeat.yml
          subPath: filebeat.yml
        * name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        * name: varlog
          mountPath: /var/log
          readOnly: true
      volumes:
      * name: config
        configMap:
          name: filebeat-config
      * name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      * name: varlog
        hostPath:
          path: /var/log
EOF
 
## Verify Filebeat is running on every node
kubectl get pods -n logging -l app=filebeat
## Expected: One pod per node, all Running

Step 4: Deploy Logstash to Process and Route Logs

Bash

## Create logstash-pipeline ConfigMap
kubectl apply -n logging -f - <<EOF
apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-pipeline
  namespace: logging
data:
  logstash.conf: |
    input {
      kafka {
        bootstrap_servers => "kafka.logging.svc.cluster.local:9092"
        topics => ["kubernetes-logs"]
        group_id => "logstash-consumer"
        codec => json
        decorate_events => true
      }
    }
 
    filter {
      # Parse the Kubernetes metadata
      if [kubernetes] {
        mutate {
          add_field => {
            "service"   => "%{[kubernetes][labels][app]}"
            "namespace" => "%{[kubernetes][namespace]}"
            "pod"       => "%{[kubernetes][pod][name]}"
          }
        }
      }
 
      # Parse JSON application logs if they are structured
      if [message] =~ /^{/ {
        json {
          source => "message"
          target => "parsed"
        }
        mutate {
          add_field => {
            "level"   => "%{[parsed][level]}"
            "msg"     => "%{[parsed][msg]}"
          }
        }
      }
 
      # Drop health check logs — they create noise
      if [message] =~ /GET \/health/ {
        drop { }
      }
 
      # Add timestamp for indexing
      date {
        match => ["timestamp", "ISO8601"]
        target => "@timestamp"
      }
    }
 
    output {
      elasticsearch {
        hosts => ["http://elasticsearch-master.logging.svc.cluster.local:9200"]
        index => "kubernetes-logs-%{+YYYY.MM.dd}"  # Daily indices for easy cleanup
        manage_template => false
      }
    }
EOF
 
## Deploy Logstash
helm install logstash elastic/logstash \
  --namespace logging \
  --set persistence.enabled=true \
  --set volumeMounts[0].name=pipeline \
  --set volumeMounts[0].mountPath=/usr/share/logstash/pipeline \
  --set volumes[0].name=pipeline \
  --set volumes[0].configMap.name=logstash-pipeline

Step 5: Create Kibana Dashboards and Watcher Alerts

Bash

## Access Kibana
kubectl port-forward -n logging svc/kibana-kibana 5601:5601
## Open http://localhost:5601

In Kibana UI:

Go to Stack Management -> Index Patterns -> Create index pattern: kubernetes-logs-*
Set time field to @timestamp
Go to Discover — you should see logs flowing in from all pods
Create a dashboard: go to Dashboard -> Create -> Add panels

Create a Watcher alert for high error rates:

Bash

## Create an alert that fires when error rate exceeds 10% in 5 minutes
curl -X PUT "http://localhost:5601/api/alerting/rule" \
  -H 'kbn-xsrf: true' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "High Error Rate Alert",
  "rule_type_id": "metrics.alert.threshold",
  "schedule": { "interval": "1m" },
  "consumer": "alerts",
  "params": {
    "criteria": [{
      "aggType": "count",
      "comparator": ">",
      "threshold": [100],
      "timeSize": 5,
      "timeUnit": "m",
      "metric": "level",
      "filterQuery": "level:ERROR"
    }]
  },
  "actions": []
}'

Validation & Testing

Bash

## 1. Verify Kafka is receiving messages
kubectl exec -n logging kafka-0 -- \
  kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --group logstash-consumer
## Expected: Shows LAG value — how many messages Logstash is behind
## LAG should be small and decreasing
 
## 2. Verify logs are in Elasticsearch
kubectl exec -n logging elasticsearch-master-0 -- \
  curl -s http://localhost:9200/_cat/indices?v | grep kubernetes-logs
## Expected: Index with today's date, doc.count increasing
 
## 3. Generate test logs
kubectl run log-test --image=busybox --rm -it --restart=Never -- \
  sh -c 'for i in $(seq 1 100); do echo "{\"level\":\"ERROR\",\"msg\":\"test error $i\",\"service\":\"test\"}"; done'
 
## 4. Search for the test logs in Kibana Discover
## Query: kubernetes.pod.name: log-test AND level: ERROR
## Expected: 100 error messages appear within 30 seconds
 
## 5. Simulate a traffic spike to test Kafka buffering
## Deploy a pod that generates 10000 log lines per second
## Watch Kafka lag increase then drain as Logstash processes the backlog
## Elasticsearch should remain healthy throughout
 
## 6. Verify log retention works
## Check that indices older than your ILM policy are being deleted
kubectl exec -n logging elasticsearch-master-0 -- \
  curl -s http://localhost:9200/_cat/indices?v | grep kubernetes-logs | sort
## Expected: Only indices from last N days (per your ILM policy)
echo "Log aggregation pipeline fully operational"

Videos & Guides

Video Guide

Elastic Stack on Kubernetes — Complete ELK Tutorial

Complete ELK stack deployment on Kubernetes covering Elasticsearch, Logstash, Kibana, Filebeat DaemonSet, and Kibana dashboard creation for production log management.

View Reference

DOCUMENTATION

Strimzi Kafka Operator Official Documentation

Official Strimzi documentation for deploying production Kafka on Kubernetes — cluster configuration, topic operators, Kafka Connect, user management, and security.

View Reference

DOCUMENTATION

Kafka Connect JDBC Connector Documentation

Official Confluent JDBC Sink Connector documentation for automatically writing Kafka topic data to PostgreSQL — upsert modes, auto table creation, and schema evolution.

View Reference