Log Pipeline with ELK and Kafka on Kubernetes

Build a Kafka-buffered log pipeline on Kubernetes with Filebeat, Logstash, Elasticsearch and Kibana, plus a z-score detector for error-rate spikes.

Architecture Overview & Problem Statement

This project builds a centralised logging pipeline on Kubernetes: Filebeat ships container logs to a Kafka buffer, Logstash parses them into Elasticsearch, Kibana searches them, and a small Python detector flags error-rate spikes. Kafka absorbs log bursts so Elasticsearch is not overloaded, and you prove that yourself with a spike test.

flowchart TD A[Pods write logs]:::blue --> B[Filebeat DaemonSet]:::blue B --> C[Kafka buffer]:::orange C --> D[Logstash parse]:::purple D --> E[Elasticsearch store]:::green E --> F[Kibana search]:::slate E --> G[Python z-score detector]:::red G -->|anomalies| E classDef blue fill:#326CE5,color:#fff classDef green fill:#10B981,color:#fff classDef orange fill:#F97316,color:#fff classDef red fill:#EF4444,color:#fff classDef purple fill:#8B5CF6,color:#fff classDef slate fill:#475569,color:#fff

What You Build, Learn and Show

You build a centralised logging pipeline on Kubernetes. Filebeat collects container logs on every node, Kafka buffers them, Logstash parses them, Elasticsearch stores them, and Kibana searches them. On top of that you add a small Python detector that watches the error rate per minute and records an anomaly when it jumps far above its recent normal.

You learn:

Why logs need to live outside the pod, and why a buffer sits between collectors and the search store
How to run Elasticsearch, Kibana, Logstash and Kafka with Kubernetes operators (ECK and Strimzi)
How to parse structured logs, keep noise out, and expire old data with index lifecycle management
How to separate a fixed-threshold alert from a baseline-aware anomaly check, and when each one is the better tool

You can show a repository with all manifests, a screenshot of Kibana Discover and your dashboard, a screenshot of the Kafka consumer lag rising and draining during your spike test, the anomaly document your detector wrote, and a short write-up with the numbers you measured on your own cluster.

Tip
this project is about log pipelines and detection logic. If you want to learn Kafka operations in depth (partitions, rebalancing, tuning), do the Kafka on Kubernetes with Strimzi project. If you want metrics-based SLO alerting, do the Prometheus SLO project.

Prerequisites

A Kubernetes cluster with at least 3 worker nodes of about 4 vCPU and 16 GB RAM each (for example 3 x t3.xlarge on EKS), or a local cluster with at least 8 CPUs and 16 GB RAM using the lab sizing in the table below
kubectl and helm 3 installed
Basic Kubernetes knowledge: Deployments, DaemonSets, Services, Secrets, PersistentVolumeClaims
Python 3 knowledge is helpful for Milestone 7, but the script is provided

For lab sizing on local clusters, the manifests below use production-style numbers. For a laptop cluster change these values and accept that you lose real high availability:

Component	Production-style (as written)	Lab sizing
Elasticsearch `count`	3	1
Kafka node pool `replicas`	3	1 (also set topic `replicas: 1` and remove `min.insync.replicas`)
Logstash `count`	2	1
Elasticsearch memory limit	4Gi	2Gi

How the Pipeline Fits Together

Every pod writes logs, but a pod's logs disappear when the pod is deleted or its node fails. That is the worst moment to lose them, because it is exactly when you need them for debugging. This pipeline copies every log line off the node as it is written.

Kafka sits between the collectors and Elasticsearch. When a service suddenly writes far more logs than usual (a retry storm, a traffic peak, an incident), Elasticsearch can fall behind or reject writes. Kafka absorbs the burst on disk, and Logstash drains it at the pace Elasticsearch can sustain.

Both stateful systems are managed by operators, which is how teams run this stack in practice:

ECK (Elastic Cloud on Kubernetes) manages Elasticsearch, Kibana, Logstash and Filebeat. It also creates TLS certificates and the elastic user password for you.
Strimzi manages Kafka in KRaft mode, so there is no ZooKeeper to run.

Tip
operator and stack versions move quickly. The versions below were chosen to work together. If you upgrade, read the ECK and Strimzi release notes first and change the version fields together.

Problem Solved

Without a log pipeline, debugging means kubectl logs pod-name and hoping the pod still exists, the log has not rotated, and you can scan thousands of lines by eye. With 50 pods you cannot search across them at all.

With this pipeline every log line from every application pod is searchable within seconds of being written. A query such as level : "ERROR" and service : "payments" across the last hour returns results from all replicas at once.

The buffer solves a second problem. During a log burst, writing straight to Elasticsearch can overload it. With Kafka in the middle you can watch the consumer lag grow while the burst lasts and shrink afterwards, and you lose nothing as long as the burst fits in Kafka retention. You will prove this yourself in the validation section rather than take it on trust.

The detector solves a third problem. A fixed rule such as "more than 1000 errors in 5 minutes" is blind to a service whose normal is 5 errors a minute doing 200 a minute, and noisy for a service whose normal is already high. A baseline-aware check compares the latest minute with that service's own recent history.

Milestone 1: Install the ECK and Strimzi Operators

This milestone installs the two operators that run Elasticsearch, Kibana, Logstash, Filebeat and Kafka for you.

Bash

## Create the namespace for the pipeline
kubectl create namespace logging
 
## ECK operator (Elasticsearch, Kibana, Logstash, Beats)
helm repo add elastic https://helm.elastic.co
helm repo update
helm install elastic-operator elastic/eck-operator \
  --namespace elastic-system --create-namespace
 
## Strimzi operator (Kafka). Watches the logging namespace.
helm install strimzi-operator \
  oci://quay.io/strimzi-helm/strimzi-kafka-operator \
  --namespace logging \
  --set watchNamespaces="{logging}"
 
## Wait until both operators are running
kubectl get pods -n elastic-system
kubectl get pods -n logging
## Expected: elastic-operator-0 and strimzi-cluster-operator-... both Running

Tip
an operator is a controller that knows how to run one piece of software. You describe the cluster you want in a custom resource, and the operator creates the pods, volumes, certificates and Services for you.

Milestone 2: Deploy Elasticsearch and Kibana

This milestone creates a three-node Elasticsearch cluster and Kibana through ECK, then sets up retention and field mappings before any logs arrive.

YAML

# elasticsearch.yaml
apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: logs
  namespace: logging
spec:
  version: 8.15.3
  nodeSets:
  - name: default
    count: 3          # 3 nodes: data stays available if one node is lost
    podTemplate:
      spec:
        containers:
        - name: elasticsearch
          resources:
            requests:
              cpu: 500m
              memory: 4Gi
            limits:
              cpu: 2
              memory: 4Gi   # JVM heap is set to half of this automatically
    volumeClaimTemplates:
    - metadata:
        name: elasticsearch-data   # this exact name is required by ECK
      spec:
        accessModes: [ReadWriteOnce]
        resources:
          requests:
            storage: 50Gi
        # storageClassName: gp3     # uncomment on EKS if you created a gp3 class
---
apiVersion: kibana.k8s.elastic.co/v1
kind: Kibana
metadata:
  name: logs
  namespace: logging
spec:
  version: 8.15.3
  count: 1
  elasticsearchRef:
    name: logs

Bash

kubectl apply -f elasticsearch.yaml
 
## Watch until HEALTH shows green (takes 3-5 minutes)
kubectl get elasticsearch,kibana -n logging --watch
 
## Get the password of the built-in "elastic" user
export ES_PASSWORD=$(kubectl get secret logs-es-elastic-user -n logging \
  -o go-template='{{.data.elastic | base64decode}}')
 
## Open a tunnel to Elasticsearch in a second terminal
kubectl port-forward -n logging svc/logs-es-http 9200:9200

Tip
never set the Elasticsearch heap above about 31 GB. Beyond that the JVM stops using compressed object pointers and performance gets worse, not better.

Security
ECK enables TLS and authentication by default. The commands below use curl -k because you are talking to a self-signed certificate through a port-forward in a lab. In a real environment, use the CA from the logs-es-http-certs-public secret and a dedicated user instead of elastic.

Now create the retention policy and index template before any logs arrive. The template makes level and service exact-match fields, and attaches the lifecycle policy so old daily indices delete themselves.

Bash

## Delete log indices 7 days after creation
curl -k -u "elastic:${ES_PASSWORD}" -X PUT \
  "https://localhost:9200/_ilm/policy/kubernetes-logs-retention" \
  -H 'Content-Type: application/json' -d '{
  "policy": { "phases": {
    "delete": { "min_age": "7d", "actions": { "delete": {} } }
  } }
}'
 
## Template applied to every daily index
curl -k -u "elastic:${ES_PASSWORD}" -X PUT \
  "https://localhost:9200/_index_template/kubernetes-logs" \
  -H 'Content-Type: application/json' -d '{
  "index_patterns": ["kubernetes-logs-*"],
  "template": {
    "settings": {
      "index.lifecycle.name": "kubernetes-logs-retention",
      "number_of_shards": 1,
      "number_of_replicas": 1
    },
    "mappings": { "properties": {
      "level":   { "type": "keyword" },
      "service": { "type": "keyword" }
    } }
  }
}'
## Expected for both: {"acknowledged":true}

Milestone 3: Deploy Kafka as the Log Buffer

This milestone creates a three-broker Kafka cluster in KRaft mode and a six-partition topic that holds logs between Filebeat and Logstash.

YAML

# kafka.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: dual-role
  namespace: logging
  labels:
    strimzi.io/cluster: logs
spec:
  replicas: 3
  roles: [controller, broker]   # KRaft: the same nodes store metadata and data
  resources:
    requests:
      cpu: 500m
      memory: 2Gi
    limits:
      cpu: 2
      memory: 2Gi
  storage:
    type: jbod
    volumes:
    - id: 0
      type: persistent-claim
      size: 20Gi
      deleteClaim: true         # lab setting: volumes are removed with the cluster
---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: logs
  namespace: logging
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    listeners:
    - name: plain
      port: 9092
      type: internal
      tls: false                # lab only: in-cluster traffic without TLS
    config:
      default.replication.factor: 3
      min.insync.replicas: 2
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
  entityOperator:
    topicOperator: {}
---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: kubernetes-logs
  namespace: logging
  labels:
    strimzi.io/cluster: logs
spec:
  partitions: 6                 # lets up to 6 Logstash consumers read in parallel
  replicas: 3
  config:
    retention.ms: 86400000      # keep 24 hours of logs in Kafka
    retention.bytes: 5368709120 # cap at 5 GiB per partition

Bash

kubectl apply -f kafka.yaml
 
## Wait for the cluster to report Ready (3-5 minutes)
kubectl wait kafka/logs -n logging --for=condition=Ready --timeout=600s
 
kubectl get pods -n logging -l strimzi.io/cluster=logs
## Expected: logs-dual-role-0, -1, -2 Running, plus the entity operator
 
kubectl get kafkatopic kubernetes-logs -n logging
## Expected: PARTITIONS 6, REPLICATION FACTOR 3, READY True

Milestone 4: Ship Container Logs with Filebeat

This milestone runs Filebeat on every node to read container logs and publish them to Kafka.

Filebeat runs one pod on every node, reads the container log files that the container runtime writes under /var/log, adds Kubernetes metadata (namespace, pod, container), and publishes each line to Kafka.

YAML

# filebeat.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: filebeat
  namespace: logging
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: filebeat
rules:
- apiGroups: [""]
  resources: [namespaces, pods, nodes]
  verbs: [get, list, watch]
- apiGroups: ["apps"]
  resources: [replicasets]
  verbs: [get, list, watch]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: filebeat
subjects:
- kind: ServiceAccount
  name: filebeat
  namespace: logging
roleRef:
  kind: ClusterRole
  name: filebeat
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: beat.k8s.elastic.co/v1beta1
kind: Beat
metadata:
  name: filebeat
  namespace: logging
spec:
  type: filebeat
  version: 8.15.3
  config:
    filebeat.inputs:
    - type: filestream
      id: kubernetes-containers
      paths:
      - /var/log/containers/*.log
      prospector.scanner.symlinks: true
      parsers:
      - container: ~              # understands the containerd/CRI log format
    processors:
    - add_kubernetes_metadata:
        host: ${NODE_NAME}
        matchers:
        - logs_path:
            logs_path: /var/log/containers/
    - drop_event:
        when:
          or:
          - equals:
              kubernetes.namespace: kube-system
          - equals:
              kubernetes.namespace: logging   # never ship the pipeline's own logs back into itself
    output.kafka:
      hosts: ["logs-kafka-bootstrap.logging.svc:9092"]
      topic: kubernetes-logs
      required_acks: 1
      compression: gzip
      max_message_bytes: 1000000
  daemonSet:
    podTemplate:
      spec:
        serviceAccountName: filebeat
        automountServiceAccountToken: true
        terminationGracePeriodSeconds: 30
        securityContext:
          runAsUser: 0            # needed to read root-owned log files on the node
        containers:
        - name: filebeat
          env:
          - name: NODE_NAME
            valueFrom:
              fieldRef:
                fieldPath: spec.nodeName
          resources:
            requests:
              cpu: 100m
              memory: 100Mi
            limits:
              cpu: 500m
              memory: 300Mi
          volumeMounts:
          - name: varlog
            mountPath: /var/log
            readOnly: true
        volumes:
        - name: varlog
          hostPath:
            path: /var/log

Bash

kubectl apply -f filebeat.yaml
 
kubectl get pods -n logging -l beat.k8s.elastic.co/name=filebeat -o wide
## Expected: one Running pod per node (compare with: kubectl get nodes)

Security
the Filebeat pod runs as root and mounts a host directory read-only. That is normal for node-level log shippers, but it is also why you should keep this DaemonSet in its own namespace and never give other workloads access to it.

Tip
this configuration assumes a containerd runtime, which is what current EKS, GKE and AKS versions use. On an old cluster that still uses Docker, you would also need to mount /var/lib/docker/containers.

Milestone 5: Parse and Route Logs with Logstash

This milestone reads logs from Kafka, parses them into searchable fields and writes them to Elasticsearch.

Logstash reads from Kafka, turns raw lines into useful fields, and writes to Elasticsearch. The elasticsearchRefs block makes ECK inject the Elasticsearch address, user, password and CA certificate as environment variables named LOGS_ES_HOSTS, LOGS_ES_USER, LOGS_ES_PASSWORD and LOGS_ES_SSL_CERTIFICATE_AUTHORITY.

YAML

# logstash.yaml
apiVersion: logstash.k8s.elastic.co/v1alpha1
kind: Logstash
metadata:
  name: logs
  namespace: logging
spec:
  count: 2                       # two consumers share the 6 partitions
  version: 8.15.3
  elasticsearchRefs:
  - clusterName: logs            # becomes the LOGS_ES_* prefix
    name: logs
  podTemplate:
    spec:
      containers:
      - name: logstash
        resources:
          requests:
            cpu: 500m
            memory: 1Gi
          limits:
            cpu: 1
            memory: 2Gi
  pipelines:
  - pipeline.id: main
    config.string: |
      input {
        kafka {
          bootstrap_servers => "logs-kafka-bootstrap.logging.svc:9092"
          topics            => ["kubernetes-logs"]
          group_id          => "logstash-consumer"
          codec             => json
        }
      }
      filter {
        # Drop health-check noise before it is stored
        if [message] =~ /GET \/health/ {
          drop { }
        }
 
        # Service name comes from the container name
        if [kubernetes][container][name] {
          mutate { add_field => { "service" => "%{[kubernetes][container][name]}" } }
        }
 
        # If the application logs JSON, lift its fields up
        if [message] =~ /^\s*\{/ {
          json {
            source => "message"
            target => "app"
            skip_on_invalid_json => true
          }
          if [app][level] {
            mutate { add_field => { "level" => "%{[app][level]}" } }
          }
          if [app][service] {
            mutate { replace => { "service" => "%{[app][service]}" } }
          }
        }
 
        # Normalise level so "error" and "ERROR" match the same query
        if [level] {
          mutate { uppercase => ["level"] }
        }
      }
 
      output {
        elasticsearch {
          hosts    => [ "${LOGS_ES_HOSTS}" ]
          user     => "${LOGS_ES_USER}"
          password => "${LOGS_ES_PASSWORD}"
          ssl_certificate_authorities => "${LOGS_ES_SSL_CERTIFICATE_AUTHORITY}"
          index        => "kubernetes-logs-%{+YYYY.MM.dd}"   # daily index, easy to expire
          manage_template => false    # we created our own template in Milestone 2
          ilm_enabled     => false    # lifecycle comes from the template
        }
      }

Bash

kubectl apply -f logstash.yaml
 
kubectl get pods -n logging -l logstash.k8s.elastic.co/name=logs
## Expected: logs-ls-0 and logs-ls-1 Running
 
kubectl logs -n logging logs-ls-0 --tail=20
## Look for "Pipeline started" and no connection errors

Milestone 6: Generate Logs and Explore Them in Kibana

This milestone adds a log-writing demo service, then lets you search the logs and build a dashboard in Kibana.

You need an application that writes logs. Deploy a small generator that writes a JSON line five times a second, with about 5% errors. It gives your detector a stable baseline to learn.

YAML

# log-generator.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: demo
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: payments
  namespace: demo
spec:
  replicas: 2
  selector:
    matchLabels:
      app: payments
  template:
    metadata:
      labels:
        app: payments
    spec:
      containers:
      - name: payments
        image: busybox:1.36
        command: ["/bin/sh", "-c"]
        args:
        - |
          i=0
          while true; do
            i=$((i+1))
            if [ $((i % 20)) -eq 0 ]; then
              echo "{\"level\":\"error\",\"service\":\"payments\",\"msg\":\"upstream timeout\"}"
            else
              echo "{\"level\":\"info\",\"service\":\"payments\",\"msg\":\"request ok\"}"
            fi
            sleep 0.2
          done
        resources:
          requests: { cpu: 10m, memory: 16Mi }
          limits:   { cpu: 50m, memory: 32Mi }

Bash

kubectl apply -f log-generator.yaml
 
## Check that documents are arriving (re-run a few times, the count should grow)
curl -k -u "elastic:${ES_PASSWORD}" \
  "https://localhost:9200/_cat/indices/kubernetes-logs-*?v"

Now open Kibana:

Bash

kubectl port-forward -n logging svc/logs-kb-http 5601:5601
## Open https://localhost:5601 (accept the self-signed certificate warning)
## Username: elastic   Password: echo $ES_PASSWORD

In the Kibana UI:

Go to Stack Management -> Data Views -> Create data view. Name: kubernetes-logs, index pattern: kubernetes-logs-*, timestamp field: @timestamp.
Open Discover. You should see logs from the payments pods.
Search with KQL: level : "ERROR" and service : "payments"
Create a dashboard (Dashboard -> Create) with three panels: log volume over time, errors over time (filter level : "ERROR"), and a top-values table on service.

Create a fixed-threshold rule for comparison:

Go to Stack Management -> Rules -> Create rule.
Choose rule type Elasticsearch query, data view kubernetes-logs, query level : "ERROR".
Condition: when the document count is above 1000 over the last 5 minutes, check every 1 minute.
Add a Server log action so you can see the alert fire without setting up email or Slack.

Milestone 7: Add the Anomaly Detector

This milestone adds a Python detector that compares each minute's error count with the last 30 minutes and records spikes.

The Kibana rule fires only when the total crosses a number you picked by hand. The detector below learns what normal looks like from the last 30 minutes and flags the latest minute when it is far outside that range. It measures how many standard deviations the latest minute is above the recent average (a z-score).

YAML

# detector.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: detector-script
  namespace: logging
data:
  detector.py: |
    import datetime
    import json
    import os
    import statistics
    import time
    import requests
 
    ES_URL = os.environ["ES_URL"]
    AUTH = ("elastic", os.environ["ES_PASSWORD"])
    CA = "/certs/ca.crt"
    Z_THRESHOLD = float(os.getenv("Z_THRESHOLD", "4"))
    MIN_ERRORS = int(os.getenv("MIN_ERRORS", "20"))   # ignore tiny absolute numbers
    BASELINE_MINUTES = 30
 
    def error_counts():
        """Return error counts per complete minute for the last hour, oldest first."""
        body = {
            "size": 0,
            "query": {"bool": {"filter": [
                {"term": {"level": "ERROR"}},
                {"range": {"@timestamp": {"gte": "now-60m/m", "lt": "now/m"}}},
            ]}},
            "aggs": {"per_minute": {"date_histogram": {
                "field": "@timestamp", "fixed_interval": "1m", "min_doc_count": 0}}},
        }
        r = requests.post(f"{ES_URL}/kubernetes-logs-*/_search", json=body,
                          auth=AUTH, verify=CA, timeout=20)
        r.raise_for_status()
        buckets = r.json()["aggregations"]["per_minute"]["buckets"]
        return [b["doc_count"] for b in buckets]
 
    def check():
        counts = error_counts()
        if len(counts) < 10:
            print("not enough history yet", len(counts), flush=True)
            return
        latest = counts[-1]
        baseline = counts[:-1][-BASELINE_MINUTES:]
        mean = statistics.mean(baseline)
        std = max(statistics.pstdev(baseline), 1.0)   # floor avoids division by ~0
        z = (latest - mean) / std
        print(json.dumps({"latest": latest, "mean": round(mean, 1),
                          "std": round(std, 1), "z": round(z, 1)}), flush=True)
        if z >= Z_THRESHOLD and latest >= MIN_ERRORS:
            doc = {
                "@timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
                "type": "error_rate_spike",
                "latest_errors_per_min": latest,
                "baseline_mean": mean,
                "baseline_std": std,
                "z_score": z,
            }
            r = requests.post(f"{ES_URL}/log-anomalies/_doc", json=doc,
                              auth=AUTH, verify=CA, timeout=20)
            r.raise_for_status()
            print("ANOMALY recorded", json.dumps(doc), flush=True)
 
    while True:
        try:
            check()
        except Exception as exc:   # keep the loop alive on transient errors
            print("detector error:", exc, flush=True)
        time.sleep(60)
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: anomaly-detector
  namespace: logging
spec:
  replicas: 1
  selector:
    matchLabels:
      app: anomaly-detector
  template:
    metadata:
      labels:
        app: anomaly-detector
    spec:
      containers:
      - name: detector
        image: python:3.12-slim
        command: ["/bin/sh", "-c"]
        args: ["pip install --no-cache-dir requests && python /app/detector.py"]
        env:
        - name: ES_URL
          value: https://logs-es-http.logging.svc:9200
        - name: ES_PASSWORD
          valueFrom:
            secretKeyRef:
              name: logs-es-elastic-user
              key: elastic
        resources:
          requests: { cpu: 50m, memory: 64Mi }
          limits:   { cpu: 200m, memory: 128Mi }
        volumeMounts:
        - name: script
          mountPath: /app
        - name: es-ca
          mountPath: /certs
          readOnly: true
      volumes:
      - name: script
        configMap:
          name: detector-script
      - name: es-ca
        secret:
          secretName: logs-es-http-certs-public

Bash

kubectl apply -f detector.yaml
 
kubectl logs -n logging deploy/anomaly-detector -f
## After a few minutes you should see lines like:
## {"latest": 30, "mean": 29.8, "std": 2.1, "z": 0.1}

Tip
a detector is only as good as its baseline. Wait at least 15 to 20 minutes after starting the generator so the baseline contains real data, otherwise the first spike is compared with almost nothing.

Validation & Testing

This final topic proves the pipeline works end to end, covers common failures and cleanup, and lists what to publish in your portfolio.

1. The pipeline delivers logs end to end

Bash

## Kafka consumer lag for Logstash (small and not growing means healthy)
kubectl exec -n logging logs-dual-role-0 -- \
  bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe --group logstash-consumer
## Expected: 6 rows (one per partition). LAG is low, usually under a few hundred.
 
## Indices exist and the doc count grows
curl -k -u "elastic:${ES_PASSWORD}" \
  "https://localhost:9200/_cat/indices/kubernetes-logs-*?v"
 
## Lifecycle policy is attached to today's index
curl -k -u "elastic:${ES_PASSWORD}" \
  "https://localhost:9200/kubernetes-logs-*/_settings?filter_path=*.settings.index.lifecycle.name"
## Expected: "name":"kubernetes-logs-retention"

In Kibana Discover, service : "payments" should return fresh documents with kubernetes.pod.name, level and service populated.

2. Logs survive a pod being deleted

Bash

## Delete one payments pod and prove its earlier logs are still searchable
POD=$(kubectl get pod -n demo -l app=payments -o name | head -1)
kubectl delete -n demo $POD

In Discover, filter on kubernetes.pod.name for the deleted pod. Its logs from before the deletion are still there. This is the core reason the project exists.

3. Spike test: Kafka absorbs a log burst

Start a burst of 20 errors per second for 3 minutes:

YAML

# error-burst.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: error-burst
  namespace: demo
spec:
  ttlSecondsAfterFinished: 300
  template:
    spec:
      restartPolicy: Never
      containers:
      - name: burst
        image: busybox:1.36
        command: ["/bin/sh", "-c"]
        args:
        - |
          end=$(( $(date +%s) + 180 ))
          while [ $(date +%s) -lt $end ]; do
            echo "{\"level\":\"error\",\"service\":\"payments\",\"msg\":\"upstream timeout\"}"
            sleep 0.05
          done

Bash

kubectl apply -f error-burst.yaml
 
## In a second terminal, watch lag every 10 seconds while the burst runs
watch -n 10 "kubectl exec -n logging logs-dual-role-0 -- \
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group logstash-consumer"

Record what you see: the peak lag, how long it took to return to near zero after the burst ended, and whether Elasticsearch stayed green (curl -k -u elastic:$ES_PASSWORD https://localhost:9200/_cluster/health?pretty). These numbers depend on your cluster, so measure them rather than copy anyone else's.

4. The detector flags the spike and the fixed rule fires

Bash

## Detector log should show a large z-score and "ANOMALY recorded"
kubectl logs -n logging deploy/anomaly-detector --tail=20
 
## The anomaly document exists
curl -k -u "elastic:${ES_PASSWORD}" \
  "https://localhost:9200/log-anomalies/_search?pretty"

Also check Stack Management -> Rules in Kibana to see whether your fixed threshold fired, and note which one reacted first. Then answer one question in your write-up: for which kind of service would the fixed rule have missed an incident that the z-score caught, and for which would the z-score be too noisy?

5. Retention works

Bash

## ILM status for your indices (needs to run after at least one day to see deletions)
curl -k -u "elastic:${ES_PASSWORD}" \
  "https://localhost:9200/kubernetes-logs-*/_ilm/explain?filter_path=indices.*.policy,indices.*.age"

For a faster test, temporarily change min_age in the policy to 5m on a spare index and confirm it disappears (ILM checks run on a poll interval, so allow several minutes).

Common Mistakes

Common Mistake
Shipping the pipeline's own logs back into the pipeline. Filebeat reads the Logstash and Elasticsearch pods' logs, which describe every document they just handled, which Filebeat reads again. Log volume grows in a loop until the disk fills. Fix: The drop_event processor excludes the logging and kube-system namespaces. Keep that exclusion.

Common Mistake
Letting Elasticsearch guess field types. level becomes analysed text, so level : "ERROR" and exact-match term queries behave unexpectedly, and the detector query returns zero. Fix: Create the index template (Milestone 2) before the first log arrives. If indices already exist with wrong mappings, delete them and let Logstash recreate them.

Common Mistake
Setting the pod memory limit and JVM heap by hand and getting them wrong. The heap is larger than the container allows, and the kernel kills Elasticsearch with OOMKilled under load. Fix: Set only the container memory limit and let ECK size the heap to about half of it.

Common Mistake
Fewer partitions than Logstash consumers. With 2 partitions and 4 Logstash pods, two pods sit idle and lag cannot be reduced by adding more. Fix: Partition count caps parallelism. Start with more partitions than consumers (here 6 versus 2).

Common Mistake
Treating the Kafka buffer as infinite. A long Elasticsearch outage outlasts the 24 hour Kafka retention, and the oldest logs are deleted before Logstash can read them. Fix: Alert on consumer lag growth, and size retention to the longest outage you want to survive.

Troubleshooting

No indices appear in Elasticsearch

Bash

## 1. Are messages reaching Kafka?
kubectl exec -n logging logs-dual-role-0 -- \
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic kubernetes-logs --max-messages 3
## Empty? Check Filebeat: kubectl logs -n logging -l beat.k8s.elastic.co/name=filebeat --tail=30
 
## 2. Is Logstash consuming and writing?
kubectl logs -n logging logs-ls-0 --tail=50
## "401" or certificate errors -> check the elasticsearchRefs block
## "Connection refused" to Kafka -> check the bootstrap address and that the cluster is Ready

Elasticsearch pods stay Pending or crash

Bash

kubectl describe pod logs-es-default-0 -n logging | tail -20
## "Insufficient memory" -> reduce count and memory (see lab sizing) or add nodes
## "unbound PersistentVolumeClaims" -> check kubectl get storageclass
## max virtual memory areas vm.max_map_count too low -> ECK sets this via an init container;
##   on locked-down clusters add node.store.allow_mmap: false to the nodeSet config (lab only)

Cluster health is yellow

With a single node or when replicas cannot be placed, indices ask for a replica that has nowhere to go. In the lab sizing, set number_of_replicas to 0 in the index template.

Detector never records an anomaly

Check its log first. not enough history yet means wait longer. If the z-score is high but nothing is recorded, check MIN_ERRORS. If the z-score stays near zero during the burst, confirm in Discover that the burst documents have level : "ERROR" (uppercase after Logstash) and are indexed.

Cost and Cleanup

This project runs about 8 pods that each reserve real memory plus roughly 270 GB of volumes. On a cloud cluster that is a meaningful hourly cost, so do not leave it running overnight.

Bash

kubectl delete -f detector.yaml -f error-burst.yaml -f log-generator.yaml --ignore-not-found
kubectl delete -f logstash.yaml -f filebeat.yaml -f kafka.yaml -f elasticsearch.yaml
helm uninstall strimzi-operator -n logging
helm uninstall elastic-operator -n elastic-system
kubectl delete pvc --all -n logging
kubectl delete namespace logging demo elastic-system
 
## Confirm no leftover volumes are still billing
kubectl get pv

Your Portfolio Deliverable

Put these in a Git repository named k8s-log-pipeline-elk-kafka:

All manifests from this guide, in a manifests/ folder
README.md with the architecture diagram, how to deploy, and how to clean up
Screenshots: Kibana Discover with parsed fields, your dashboard, and the Kafka lag during the spike test
RESULTS.md with your measured numbers: peak lag, time to drain, Elasticsearch health during the burst, the z-score that triggered the anomaly, and which alert reacted first
A short "Decisions and trade-offs" section: why Kafka sits in the middle, why partition count is 6, why you excluded certain namespaces, and what you would change before running this for real

Stretch Goals

Add a per-service detector by grouping the aggregation with a terms sub-aggregation on service
Replace the plain Kafka listener with TLS and SCRAM authentication in Strimzi
Send the detector's anomaly documents to Slack or PagerDuty using a Kibana rule on the log-anomalies index
Compare your z-score detector with Elasticsearch's built-in machine learning jobs (they need a paid licence or a trial) and write down when each one is worth it

Found a mistake, an outdated step or a broken link?

Log Pipeline with ELK and Kafka on Kubernetes

Blueprint Walkthrough

Architecture Overview & Problem Statement

What You Build, Learn and Show

Prerequisites

How the Pipeline Fits Together

Problem Solved

Milestone 1: Install the ECK and Strimzi Operators

Milestone 2: Deploy Elasticsearch and Kibana

Milestone 3: Deploy Kafka as the Log Buffer

Milestone 4: Ship Container Logs with Filebeat

Milestone 5: Parse and Route Logs with Logstash

Milestone 6: Generate Logs and Explore Them in Kibana

Milestone 7: Add the Anomaly Detector

Validation & Testing

1. The pipeline delivers logs end to end

2. Logs survive a pod being deleted

3. Spike test: Kafka absorbs a log burst

4. The detector flags the spike and the fixed rule fires

5. Retention works

Common Mistakes

Troubleshooting

No indices appear in Elasticsearch

Elasticsearch pods stay Pending or crash

Cluster health is yellow

Detector never records an anomaly

Cost and Cleanup

Your Portfolio Deliverable

Stretch Goals