Logging with ELK Stack

Centralized logging enables collecting, storing, and analyzing logs from all systems in one place for troubleshooting and compliance.

Why Centralized Logging?

Distributed Systems Problem:
┌─────────┐  ┌─────────┐  ┌─────────┐
│  App 1  │  │  App 2  │  │  App 3  │
│  Logs   │  │  Logs   │  │  Logs   │
└────┬────┘  └────┬────┘  └────┬────┘
     │            │            │
     ▼            ▼            ▼
  Multiple files to check... HARD!

With ELK:
┌─────────┐  ┌─────────┐  ┌─────────┐
│  App 1  │  │  App 2  │  │  App 3  │
└────┬────┘  └────┬────┘  └────┬────┘
     │            │            │
     └────────────┼────────────┘
                  ▼
          ┌─────────────┐
          │   Beats     │  (Log shippers)
          └──────┬──────┘
                 ▼
          ┌─────────────┐
          │  Logstash   │  (Parse/enrich)
          └──────┬──────┘
                 ▼
          ┌─────────────┐
          │Elasticsearch│  (Store/index)
          └──────┬──────┘
                 ▼
          ┌─────────────┐
          │   Kibana    │  (Visualize)
          └─────────────┘
     Single interface to search ALL logs!

ELK Stack Components

Filebeat

Lightweight log shipper.

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/*.log
    - /var/log/app/*.log
  fields:
    service: nginx
    environment: production
  fields_under_root: true
  multiline.pattern: '^\['
  multiline.negate: true
  multiline.match: after

- type: docker
  containers.ids: '*'
  containers.path: "/var/lib/docker/containers"
  containers.stream: all

processors:
- add_host_metadata:
    when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~

output.elasticsearch:
  hosts: ["https://elasticsearch:9200"]
  index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"
  username: "${ES_USERNAME}"
  password: "${ES_PASSWORD}"
  ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]

# Or send to Logstash
# output.logstash:
#   hosts: ["logstash:5044"]

logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

# Kubernetes Filebeat DaemonSet
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: ELASTICSEARCH_HOST
          value: elasticsearch
        - name: ELASTICSEARCH_PORT
          value: "9200"
        - name: ELASTICSEARCH_USERNAME
          value: elastic
        - name: ELASTICSEARCH_PASSWORD
          valueFrom:
            secretKeyRef:
              name: elasticsearch
              key: password
        securityContext:
          runAsUser: 0
          capabilities:
            add:
            - DAC_READ_SEARCH
        volumeMounts:
        - name: config
          mountPath: /etc/filebeat.yml
          readOnly: true
          subPath: filebeat.yml
        - name: data
          mountPath: /usr/share/filebeat/data
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        - name: varlog
          mountPath: /var/log
          readOnly: true
      volumes:
      - name: config
        configMap:
          name: filebeat-config
      - name: data
          hostPath:
            path: /var/lib/filebeat-data
            type: DirectoryOrCreate
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: varlog
        hostPath:
          path: /var/log

Logstash

Data processing pipeline.

# logstash.conf
input {
  beats {
    port => 5044
  }
  
  tcp {
    port => 5000
    codec => json
  }
}

filter {
  # Grok parsing for nginx logs
  if [fields][service] == "nginx" {
    grok {
      match => {
        "message" => '%{IPORHOST:client_ip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status} %{NUMBER:bytes} "%{DATA:referrer}" "%{DATA:user_agent}"'
      }
    }
    
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
    
    geoip {
      source => "client_ip"
      target => "geoip"
    }
    
    mutate {
      convert => {
        "status" => "integer"
        "bytes" => "integer"
      }
      remove_field => ["timestamp", "beat", "input", "host", "agent", "ecs", "log", "tags"]
    }
  }
  
  # Application logs
  if [fields][service] == "app" {
    json {
      source => "message"
      target => "parsed"
    }
    
    if [parsed][level] == "ERROR" {
      mutate {
        add_tag => ["error_log"]
      }
    }
  }
  
  # Drop debug logs in production
  if [fields][environment] == "production" and [level] == "DEBUG" {
    drop {}
  }
}

output {
  if "error_log" in [tags] {
    elasticsearch {
      hosts => ["${ES_HOST:elasticsearch:9200}"]
      index => "error-logs-%{+yyyy.MM.dd}"
      user => "${ES_USERNAME}"
      password => "${ES_PASSWORD}"
    }
  } else {
    elasticsearch {
      hosts => ["${ES_HOST:elasticsearch:9200}"]
      index => "%{[fields][service]}-%{[fields][environment]}-%{+yyyy.MM.dd}"
      user => "${ES_USERNAME}"
      password => "${ES_PASSWORD}"
    }
  }
  
  # Also output to stdout for debugging
  stdout { codec => rubydebug }
}

Elasticsearch

Search and analytics engine.

# elasticsearch.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: elasticsearch
  namespace: logging
spec:
  serviceName: elasticsearch
  replicas: 3
  selector:
    matchLabels:
      app: elasticsearch
  template:
    metadata:
      labels:
        app: elasticsearch
    spec:
      containers:
      - name: elasticsearch
        image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
        resources:
          limits:
            cpu: 1000m
            memory: 2Gi
          requests:
            cpu: 100m
            memory: 1Gi
        ports:
        - containerPort: 9200
          name: rest
          protocol: TCP
        - containerPort: 9300
          name: inter-node
          protocol: TCP
        volumeMounts:
        - name: data
          mountPath: /usr/share/elasticsearch/data
        env:
        - name: cluster.name
          value: elasticsearch-cluster
        - name: node.name
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: discovery.seed_hosts
          value: "elasticsearch-0.elasticsearch,elasticsearch-1.elasticsearch,elasticsearch-2.elasticsearch"
        - name: cluster.initial_master_nodes
          value: "elasticsearch-0,elasticsearch-1,elasticsearch-2"
        - name: ES_JAVA_OPTS
          value: "-Xms512m -Xmx512m"
        - name: xpack.security.enabled
          value: "false"
  volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: standard
      resources:
        requests:
          storage: 10Gi

Index Lifecycle Management

PUT _ilm/policy/logs_policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_primary_shard_size": "50gb",
            "max_age": "1d",
            "max_docs": 100000000
          }
        }
      },
      "warm": {
        "min_age": "2d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      },
      "cold": {
        "min_age": "7d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "30d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

Kibana

Visualization and exploration.

# kibana.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kibana
  namespace: logging
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kibana
  template:
    metadata:
      labels:
        app: kibana
    spec:
      containers:
      - name: kibana
        image: docker.elastic.co/kibana/kibana:8.11.0
        ports:
        - containerPort: 5601
        env:
        - name: ELASTICSEARCH_HOSTS
          value: '["http://elasticsearch:9200"]'
        - name: SERVER_NAME
          value: kibana
        - name: XPACK_SECURITY_ENABLED
          value: "false"
        resources:
          limits:
            cpu: 1000m
            memory: 1Gi
          requests:
            cpu: 100m
            memory: 512Mi

Kibana Search Queries

# Basic search
status:500

# Range queries
@timestamp:[now-1h TO now]
status >= 400 AND status < 500

# Boolean queries
status:200 AND method:POST
status:(400 OR 500)
NOT status:200

# Wildcards
user_agent:*Chrome*
message:error*

# Field existence
_exists_:geoip.country_name

# Regex
message:/error.*database/

Logging Best Practices

Structured Logging

{
  "timestamp": "2024-01-15T10:30:00Z",
  "level": "ERROR",
  "service": "payment-service",
  "environment": "production",
  "trace_id": "abc123",
  "span_id": "def456",
  "user_id": "user789",
  "message": "Payment processing failed",
  "error": {
    "type": "PaymentGatewayError",
    "code": "CARD_DECLINED",
    "message": "Insufficient funds"
  },
  "context": {
    "amount": 99.99,
    "currency": "USD",
    "merchant_id": "merch123"
  },
  "source": {
    "file": "payment.go",
    "line": 156,
    "function": "ProcessPayment"
  }
}

Log Levels

Quiz

Next Steps

Now let's explore DevSecOps—integrating security into the DevOps pipeline.