Centralized Log Management with ELK Stack on Linux

Introduction

Centralized log management is critical for troubleshooting, security auditing, and compliance. The ELK Stack (Elasticsearch, Logstash, Kibana) combined with Filebeat/Metricbeat provides a complete log management solution for Linux infrastructure.

Why Centralized Logging?

Logs disappear when a server fails — centralized storage survives failures
Grep across 50 servers simultaneously
Correlate events across systems (web request → app log → DB log)
Meet audit and compliance requirements (PCI-DSS, ISO 27001)
Detect security incidents by correlating multiple log sources

Log Sources to Collect

Source	File/Method	Priority
Auth/SSH	/var/log/auth.log	Critical
Syslog	/var/log/syslog	High
Nginx/Apache	/var/log/nginx/*.log	High
Application	/var/log/myapp/*.log	High
Kernel	/var/log/kern.log	Medium
Cron	/var/log/cron.log	Low

Filebeat Configuration (on each Linux server)

Install Filebeat:

BASH

apt install filebeat

Configure /etc/filebeat/filebeat.yml:

YAML

filebeat.inputs:

- type: log
  enabled: true
  paths:
    - /var/log/auth.log
  fields:
    log_type: auth
    environment: production
    hostname: ${HOSTNAME}

- type: log
  enabled: true
  paths:
    - /var/log/syslog
  fields:
    log_type: syslog

- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    log_type: nginx_access
  json.keys_under_root: true
  json.add_error_key: true

- type: log
  enabled: true
  paths:
    - /var/log/nginx/error.log
  fields:
    log_type: nginx_error

# Multiline for Java stack traces
- type: log
  enabled: true
  paths:
    - /var/log/myapp/*.log
  fields:
    log_type: application
  multiline:
    pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
    negate: true
    match: after

output.logstash:
  hosts: ["logstash-server:5044"]
  ssl.certificate_authorities: ["/etc/filebeat/certs/ca.crt"]
  ssl.certificate: "/etc/filebeat/certs/filebeat.crt"
  ssl.key: "/etc/filebeat/certs/filebeat.key"

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - drop_fields:
      fields: ["agent.ephemeral_id"]

Logstash Parsing Pipelines

Create /etc/logstash/conf.d/auth.conf:

RUBY

filter {
  if [fields][log_type] == "auth" {
    grok {
      match => {
        "message" => [
          # SSH failed login
          "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:host} sshd[%{POSINT:pid}]: Failed password for %{USER:failed_user} from %{IP:src_ip} port %{INT:src_port}",
          # SSH successful login
          "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:host} sshd[%{POSINT:pid}]: Accepted publickey for %{USER:auth_user} from %{IP:src_ip}",
          # sudo
          "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:host} sudo:s+%{USER:sudo_user} : TTY=%{DATA:tty} ; PWD=%{PATH:pwd} ; USER=%{USER:run_as_user} ; COMMAND=%{GREEDYDATA:command}"
        ]
      }
    }
    
    # GeoIP lookup for source IP
    if [src_ip] {
      geoip {
        source => "src_ip"
        target => "geoip"
      }
    }
    
    date {
      match => ["timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
    }
  }
}

Create /etc/logstash/conf.d/nginx.conf:

RUBY

filter {
  if [fields][log_type] == "nginx_access" {
    grok {
      match => {
        "message" => '%{IPORHOST:client_ip} - %{DATA:user} [%{HTTPDATE:timestamp}] "%{WORD:method} %{DATA:request_uri} HTTP/%{NUMBER:http_version}" %{INT:status_code} %{INT:bytes} "%{DATA:referrer}" "%{DATA:user_agent}"'
      }
    }
    
    mutate {
      convert => {
        "status_code" => "integer"
        "bytes" => "integer"
      }
    }
    
    geoip {
      source => "client_ip"
    }
    
    useragent {
      source => "user_agent"
      target => "ua"
    }
  }
}

Security Dashboards in Kibana

Key visualizations to build:

TEXT

1. SSH Failed Logins Map
   - World map showing src_ip GeoIP locations for failed SSH
   - Filter: log_type:auth AND message:"Failed password"

2. HTTP Error Rate
   - Line chart of 4xx/5xx over time
   - Filter: log_type:nginx_access AND status_code >= 400

3. Top Attackers
   - Data table: top src_ip by count for failed SSH
   - Alert if single IP fails 10+ times in 5 minutes

4. Successful Sudo Commands
   - Who ran what as root
   - Filter: log_type:auth AND message:"sudo"

Log Retention with ILM

Configure Index Lifecycle Management:

BASH

# Hot: 0-7 days (active writing)
# Warm: 7-30 days (read only, compressed)
# Delete: after 90 days

curl -X PUT "localhost:9200/_ilm/policy/logs-policy" -H 'Content-Type: application/json' -d'
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {"max_age": "1d", "max_size": "50gb"},
          "set_priority": {"priority": 100}
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "allocate": {"number_of_replicas": 0},
          "shrink": {"number_of_shards": 1},
          "forcemerge": {"max_num_segments": 1},
          "set_priority": {"priority": 50}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {"delete": {}}
      }
    }
  }
}'