log-aggregation

Log Aggregation (ELK Stack)

When to Use

• Deploying or configuring Elasticsearch clusters
• Building Logstash pipelines (inputs, filters, outputs)
• Configuring Filebeat/Metricbeat/Heartbeat shippers
• Setting up Kibana dashboards and alerts
• Debugging log flow from source to Elasticsearch
• Diagnosing cluster health (red/yellow status)

Elasticsearch

Cluster Health

# Overall cluster status
curl -s localhost:9200/_cluster/health?pretty

# Node-level stats
curl -s localhost:9200/_nodes/stats?pretty

# Shard allocation explanation (when yellow/red)
curl -s localhost:9200/_cluster/allocation/explain?pretty

# Index-level health
curl -s localhost:9200/_cat/indices?v&health=yellow
curl -s localhost:9200/_cat/indices?v&health=red

Index Management

# List indices with sizes
curl -s localhost:9200/_cat/indices?v&h=index,docs.count,store.size,pri,rep

# Create index with mapping
curl -X PUT localhost:9200/my-index -H 'Content-Type: application/json' -d '
{
  "mappings": {
    "properties": {
      "@timestamp": { "type": "date" },
      "message": { "type": "text" },
      "host": { "type": "keyword" },
      "level": { "type": "keyword" }
    }
  }
}'

# Delete index
curl -X DELETE localhost:9200/my-index

# Force merge (reduce segments)
curl -X POST "localhost:9200/my-index/_forcemerge?max_num_segments=1"

Cluster Red/Yellow Diagnosis

1. Check health: curl localhost:9200/_cluster/health?pretty
2. Check unassigned shards: curl localhost:9200/_cat/shards?v&h=index,shard,_pri,state,node&s=state
3. Get allocation explanation: curl -X POST localhost:9200/_cluster/allocation/explain?pretty
4. Common causes: disk watermark exceeded, node down, replica count > data nodes
5. Check disk: curl localhost:9200/_cat/allocation?v
6. Check settings: curl localhost:9200/_cluster/settings?include_defaults=true&flat_settings=true

Logstash

Pipeline Configuration

# /etc/logstash/pipeline/my-pipeline.conf

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["logs"]
    group_id => "logstash-consumer"
    consumer_threads => 4
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
  }
  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
  }
  mutate {
    remove_field => ["timestamp"]
  }
}

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