Chapter 7.2: Centralized Logging (ELK Stack)

(Note: A fourth component, "Beats," is often used. "Filebeat" is a lightweight *agent* you install on your servers to "ship" logs *to* Logstash. For our purposes, we'll group it with Logstash).

Part 1: Elasticsearch (The "Database")

Elasticsearch is the heart of the stack. It's built on top of **Apache Lucene** and is, at its core, a **full-text search engine**, not a traditional database. This makes it incredibly fast at searching through terabytes of unstructured text data (like logs).

Core Concepts (SQL vs. Elasticsearch)

To understand Elasticsearch, it's helpful to compare it to a SQL database:

• SQL Database -> **Elasticsearch Index** (e.g., logs-2025-11-10)
• SQL Table -> **(No direct equivalent)**
• SQL Row -> **Elasticsearch Document** (a JSON object)
• SQL Column -> **Elasticsearch Field** (a key in the JSON object)
• SQL Schema -> **Elasticsearch Mapping**

How it Works: The Inverted Index

Why is Elasticsearch so fast? It doesn't scan every log line. It uses an **Inverted Index**, just like the index at the back of a textbook.
When you save a log "user 'admin' failed", Elasticsearch *doesn't* just store the string. It breaks it down and updates its index:

• user -> (in document 1, 5, 7)
• admin -> (in document 1, 44)
• failed -> (in document 1, 2, 99)

When you *search* for "admin" AND "failed", it just finds the intersection of those two lists (document 1). This is lightning fast.

Running Elasticsearch with Docker

Elasticsearch is complex to install, but trivial with Docker.

# Run a single-node, development-mode cluster
docker run -d \
    -p 9200:9200 \
    -p 9300:9300 \
    -e "discovery.type=single-node" \
    --name elasticsearch \
    docker.elastic.co/elasticsearch/elasticsearch:8.10.4

Querying Elasticsearch (with `curl`)

Elasticsearch has a powerful REST API that you can query with `curl`. You don't use SQL; you use **Query DSL** (a JSON-based query language).

# Check the cluster health (wait for status to be 'green' or 'yellow')
$ curl -X GET "localhost:9200/_cluster/health?pretty"

# Search for *all* logs in *all* indexes
$ curl -X GET "localhost:9200/_search?pretty"

# Use Query DSL: Find all logs where 'level' is 'ERROR'
$ curl -X GET "localhost:9200/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "query": {
    "match": {
      "level": "ERROR"
    }
  }
}
'

You will almost never do this by hand. You will use **Kibana** to do this for you.

Part 2: Logstash (The "Pipeline")

Logstash is the "L" in ELK. It's a server-side data processing pipeline. Its job is to pull data from many sources, transform it, and send it to a "stash" (like Elasticsearch).

The 3 Stages of a Logstash Pipeline

You configure Logstash with a simple config file (e.g., logstash.conf) that has three sections:

1. input:** Where are the logs coming from? (e.g., a file, a port, another service).
2. filter:** How do we **parse** and **transform** the logs? (This is the most important part).
3. output:** Where do the processed logs go? (e.g., Elasticsearch, or just the console for debugging).

127.0.0.1 - - [10/Nov/2025:13:55:01 +0000] "GET /api/users HTTP/1.1" 200 456

{
  "client_ip": "127.0.0.1",
  "http_verb": "GET",
  "http_path": "/api/users",
  "http_status": 200,
  "bytes_sent": 456
}

input {
  # We will read from a file
  file {
    path => "/var/log/nginx.log"
    start_position => "beginning"
  }
}

filter {
  # This is the 'grok' filter, the heart of Logstash
  grok {
    # 'match' defines the pattern to use
    # '%{...}' is a pre-defined grok pattern
    match => { "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:http_verb} %{URIPATHPARAM:http_path} HTTP/%{NUMBER:http_version}\" %{INT:http_status} %{INT:bytes_sent}" }
  }
  
  # After grok, http_status is a "String" ("200").
  # We need to convert it to a number.
  mutate {
    convert => {
      "http_status" => "integer"
      "bytes_sent" => "integer"
    }
  }
  
  # Parse the timestamp so Elasticsearch knows it's a date
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  # 1. Send the data to our Elasticsearch container
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "nginx-logs-%{+YYYY.MM.dd}" # Create a new index every day
  }
  
  # 2. Also print it to the console for debugging
  stdout {
    codec => rubydebug
  }
}

[SERVICE]
    Log_Level    info
    Daemon       Off

[INPUT]
    Name         tail
    Path         /var/log/containers/*.log
    Parser       docker # Use the built-in docker log parser

[FILTER]
    Name         kubernetes
    Match        *
    Kube_URL     https://kubernetes.default.svc:443
    Merge_Log    On # Merge the JSON log string into the main object

[OUTPUT]
    Name         es # (Elasticsearch)
    Match        *
    Host         elasticsearch-master
    Port         9200
    Logstash_Format On
    Index        k8s-logs-%Y.%m.%d