ELK Stack ile Nginx Log Analizi: Elasticsearch, Logstash ve Kibana Kurulumu

Yazar:komut.io ekibi ELK Stack ile Log Yönetimi ve Analizi
0 Yorum

Nginx loglarına bakmak zorunda kaldığınızda ne hissediyorsunuz? Ben ilk yıllarda tail -f access.log açıp ekrana akan satırları izlerken kendimi çok akıllı sanırdım. Sonra bir gün 50 milyon satırlık log dosyasından “şu IP adresi ne zaman hangi endpoint’e kaç istek attı” sorusunu yanıtlamam gerekti. O gün ELK Stack ile ciddi anlamda tanıştım.

Bu yazıda sıfırdan bir Nginx log analiz altyapısı kuracağız. Filebeat ile logları toplayacak, Logstash ile işleyecek, Elasticsearch’e atacak ve Kibana üzerinden anlamlı dashboard’lar oluşturacağız. Adımlar gerçek bir production ortamından alındı, dolayısıyla “bu konfigürasyonu neden böyle yapıyoruz” sorularını da yanıtlamaya çalışacağım.

Ortam Hakkında Birkaç Söz

Kurulumu Ubuntu 22.04 üzerinde yapıyoruz. ELK Stack için minimum 8GB RAM öneriyorum, 16GB olursa Logstash pipeline’larında çok daha rahat edersiniz. Elasticsearch inanılmaz RAM dostu bir yazılım değil, JVM heap ayarlarını mutlaka düzgün yapmanız gerekiyor.

Kullanacağımız versiyonlar:

  • Elasticsearch: 8.x
  • Logstash: 8.x
  • Kibana: 8.x
  • Filebeat: 8.x

Versiyon uyumsuzluğu ELK Stack’in en can sıkıcı meselelerinden biri. Aynı major version kullanın, hayatınız kolaylaşır.

Elasticsearch Kurulumu

Elasticsearch kurulumuna GPG anahtarı ekleyerek başlıyoruz:

wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elasticsearch-keyring.gpg

echo "deb [signed-by=/usr/share/keyrings/elasticsearch-keyring.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list

sudo apt update && sudo apt install elasticsearch -y

Kurulum bittikten sonra JVM heap boyutunu ayarlayın. Elasticsearch’ün önerisi sisteminizin RAM’inin yarısını vermek, ama bu değeri 32GB’ın üzerine çıkarmayın. JVM’de compressed oops özelliği 32GB sınırında devre dışı kalıyor ve performans düşebiliyor.

sudo nano /etc/elasticsearch/jvm.options.d/heap.options

Dosyaya şunu ekleyin (8GB RAM için):

-Xms4g
-Xmx4g

Elasticsearch’ü başlatmadan önce elasticsearch.yml dosyasını da düzenleyin:

sudo nano /etc/elasticsearch/elasticsearch.yml
cluster.name: nginx-log-cluster
node.name: node-1
network.host: 0.0.0.0
http.port: 9200
discovery.type: single-node
xpack.security.enabled: true
xpack.security.enrollment.enabled: true

single-node modunu production’da tek sunucu çalışıyorsanız kullanıyorsunuz. Cluster kuruyorsanız bu satırı kaldırıp discovery ayarlarını yapmanız gerekiyor. Şimdilik tek node ile devam ediyoruz.

sudo systemctl enable elasticsearch
sudo systemctl start elasticsearch

# Kurulum sırasında oluşan şifreyi not alın
sudo /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic

Kibana Kurulumu

Elasticsearch ile aynı repo’dan geldiği için ek kaynak eklemenize gerek yok:

sudo apt install kibana -y
sudo nano /etc/kibana/kibana.yml
server.port: 5601
server.host: "0.0.0.0"
server.name: "nginx-log-kibana"
elasticsearch.hosts: ["http://localhost:9200"]
elasticsearch.username: "kibana_system"
elasticsearch.ssl.verificationMode: none

Kibana için kibana_system kullanıcısının şifresini sıfırlayın:

sudo /usr/share/elasticsearch/bin/elasticsearch-reset-password -u kibana_system

Bu şifreyi kibana.yml dosyasına elasticsearch.password olarak ekleyin. Şifreyi düz metin olarak dosyaya yazmak istemiyorsanız Kibana keystore kullanabilirsiniz:

sudo /usr/share/kibana/bin/kibana-keystore create
sudo /usr/share/kibana/bin/kibana-keystore add elasticsearch.password
sudo systemctl enable kibana && sudo systemctl start kibana

Logstash Kurulumu ve Pipeline Yapılandırması

Logstash, ELK’nin en güçlü ve aynı zamanda en çok kaynak tüketen parçası. Nginx loglarını parse etmek için Grok filter kullanacağız. Grok, regex’in daha okunabilir hali diyebilirsiniz, ama regex bilmeden de anlayabileceğiniz bir format.

sudo apt install logstash -y

Önce Nginx’in log formatını anlamamız gerekiyor. Varsayılan Nginx combined log formatı şöyle görünür:

192.168.1.100 - frank [10/Oct/2023:13:55:36 -0700] "GET /api/users HTTP/1.1" 200 2326 "http://example.com/" "Mozilla/5.0..."

Şimdi bu formatı parse edecek Logstash pipeline’ını oluşturalım:

sudo nano /etc/logstash/conf.d/nginx-pipeline.conf
input {
  beats {
    port => 5044
    ssl => false
  }
}

filter {
  if [fields][log_type] == "nginx_access" {
    grok {
      match => {
        "message" => '%{IPORHOST:client_ip} - %{DATA:ident} [%{HTTPDATE:timestamp}] "(?:%{WORD:http_method} %{NOTSPACE:request_uri}(?: HTTP/%{NUMBER:http_version})?|-)" %{NUMBER:response_code:int} (?:%{NUMBER:response_bytes:int}|-) "(?:%{DATA:referrer}|-)" "(?:%{DATA:user_agent}|-)"'
      }
    }

    if "_grokparsefailure" not in [tags] {
      date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
        target => "@timestamp"
        remove_field => ["timestamp"]
      }

      mutate {
        convert => {
          "response_code" => "integer"
          "response_bytes" => "integer"
        }
        add_field => {
          "environment" => "production"
        }
      }

      useragent {
        source => "user_agent"
        target => "ua"
      }

      geoip {
        source => "client_ip"
        target => "geoip"
        fields => ["city_name", "country_name", "country_code2", "location"]
      }
    }
  }

  if [fields][log_type] == "nginx_error" {
    grok {
      match => {
        "message" => '%{DATA:timestamp} [%{DATA:error_level}] %{NUMBER:pid}#%{NUMBER:tid}: (?:*%{NUMBER:connection_id} )?%{GREEDYDATA:error_message}'
      }
    }
  }
}

output {
  if [fields][log_type] == "nginx_access" {
    elasticsearch {
      hosts => ["localhost:9200"]
      user => "elastic"
      password => "YOUR_ELASTIC_PASSWORD"
      index => "nginx-access-%{+YYYY.MM.dd}"
      ssl => false
    }
  }

  if [fields][log_type] == "nginx_error" {
    elasticsearch {
      hosts => ["localhost:9200"]
      user => "elastic"
      password => "YOUR_ELASTIC_PASSWORD"
      index => "nginx-error-%{+YYYY.MM.dd}"
      ssl => false
    }
  }
}

Bu pipeline’da dikkat etmenizi istediğim birkaç nokta var. geoip filter’ı IP adreslerini coğrafi konuma dönüştürüyor. Türkiye’deki sistem yöneticileri için bu özellik çok değerli çünkü hangi ülkelerden trafik geldiğini anında görebiliyorsunuz. useragent filter’ı ise tarayıcı ve işletim sistemi bilgisini otomatik parse ediyor.

Index adlarına tarih eklemek (nginx-access-%{+YYYY.MM.dd}) ILM (Index Lifecycle Management) kullanırken şart. Eski logları otomatik silmek ya da cold storage’a taşımak için bu yapıya ihtiyacınız var.

Logstash’ı başlatmadan önce pipeline’ı test edin:

sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash -t

Test başarılıysa:

sudo systemctl enable logstash && sudo systemctl start logstash

Filebeat ile Log Toplama

Filebeat, log dosyalarını izleyip Logstash’a gönderen hafif bir shipper. Nginx sunucularına kurulacak, merkezi ELK Stack’e log gönderecek.

sudo apt install filebeat -y
sudo nano /etc/filebeat/filebeat.yml
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/nginx/access.log
      - /var/log/nginx/*.access.log
    fields:
      log_type: nginx_access
    fields_under_root: false
    multiline.pattern: '^d{1,3}.d{1,3}.d{1,3}.d{1,3}'
    multiline.negate: true
    multiline.match: after

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

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false

setup.template.settings:
  index.number_of_shards: 1

output.logstash:
  hosts: ["ELK_SERVER_IP:5044"]
  compression_level: 3

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - drop_fields:
      fields: ["agent.ephemeral_id", "ecs.version"]
      ignore_missing: true

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

fields_under_root: false seçeneğine dikkat edin. Bu ayar true olursa log_type alanı root’a çıkar ve mevcut alanları ezebilir. Logstash pipeline’ımızda [fields][log_type] şeklinde eriştiğimiz için false bırakıyoruz.

sudo systemctl enable filebeat && sudo systemctl start filebeat

# Log akışını kontrol edin
sudo journalctl -u filebeat -f

Elasticsearch Index Template Ayarları

Loglar akmaya başlamadan önce index template oluşturun. Bu adımı atlayan çoğu kişi sonradan “neden response_code text olarak kaydedilmiş” gibi sorunlarla karşılaşıyor.

curl -X PUT "localhost:9200/_index_template/nginx-access-template" 
  -u elastic:YOUR_PASSWORD 
  -H "Content-Type: application/json" 
  -d '{
    "index_patterns": ["nginx-access-*"],
    "template": {
      "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 0,
        "index.refresh_interval": "5s"
      },
      "mappings": {
        "properties": {
          "@timestamp": {"type": "date"},
          "client_ip": {"type": "ip"},
          "response_code": {"type": "integer"},
          "response_bytes": {"type": "long"},
          "request_uri": {"type": "keyword"},
          "http_method": {"type": "keyword"},
          "geoip": {
            "properties": {
              "location": {"type": "geo_point"},
              "country_name": {"type": "keyword"},
              "city_name": {"type": "keyword"}
            }
          }
        }
      }
    }
  }'

geo_point tipi Kibana’daki harita visualizasyonu için şart. client_ip alanını ip tipi olarak tanımlamak ise IP bazlı filtreleme sorgularını çok hızlandırıyor.

Kibana’da Dashboard Oluşturma

Veriler akmaya başladıktan sonra Kibana’ya gidin (http://SUNUCU_IP:5601) ve önce Data View oluşturun. Discover ekranında nginx-access-* pattern’ini ekleyin, timestamp field olarak @timestamp seçin.

Faydalı bulduğum visualizasyonlar:

  • HTTP Status Code Dağılımı: Pie chart, terms aggregation response_code üzerinde. 5xx oranı %1’i geçiyorsa alarm vermeniz gerekiyor.
  • Top 10 İstek Yapan IP’ler: Data table, client_ip terms aggregation. DDoS veya scraping tespiti için.
  • Saatlik İstek Yoğunluğu: Line chart, date histogram. Trafik örüntüsünü anlamak için.
  • Ülke Bazlı Trafik: Map visualization, geoip.location ile. Beklenmedik ülkelerden gelen trafik güvenlik açısından değerli.
  • Yavaş Endpoint’ler: Ortalama response time’a göre sıralı tablo. Bunun için Nginx log formatına $request_time eklemeniz gerekiyor.

$request_time için Nginx log formatını güncelleyin:

sudo nano /etc/nginx/nginx.conf
http {
    log_format detailed '$remote_addr - $remote_user [$time_local] '
                        '"$request" $status $body_bytes_sent '
                        '"$http_referer" "$http_user_agent" '
                        'rt=$request_time uct="$upstream_connect_time" '
                        'uht="$upstream_header_time" urt="$upstream_response_time"';

    access_log /var/log/nginx/access.log detailed;
}

Bu format değişikliğinden sonra Logstash Grok pattern’ini de güncellemeniz gerekecek:

grok {
  match => {
    "message" => '%{IPORHOST:client_ip} - %{DATA:ident} [%{HTTPDATE:timestamp}] "(?:%{WORD:http_method} %{NOTSPACE:request_uri}(?: HTTP/%{NUMBER:http_version})?|-)" %{NUMBER:response_code:int} (?:%{NUMBER:response_bytes:int}|-) "(?:%{DATA:referrer}|-)" "(?:%{DATA:user_agent}|-)" rt=%{NUMBER:request_time:float}'
  }
}

ILM ile Otomatik Log Yönetimi

Log birikmeye başladığında disk alanı sorunu kaçınılmaz. Index Lifecycle Management ile 7 günden eski access loglarını otomatik silebilirsiniz:

curl -X PUT "localhost:9200/_ilm/policy/nginx-log-policy" 
  -u elastic:YOUR_PASSWORD 
  -H "Content-Type: application/json" 
  -d '{
    "policy": {
      "phases": {
        "hot": {
          "actions": {
            "rollover": {
              "max_age": "1d",
              "max_size": "10gb"
            }
          }
        },
        "warm": {
          "min_age": "3d",
          "actions": {
            "forcemerge": {
              "max_num_segments": 1
            },
            "shrink": {
              "number_of_shards": 1
            }
          }
        },
        "delete": {
          "min_age": "30d",
          "actions": {
            "delete": {}
          }
        }
      }
    }
  }'

Bu politika ile loglar 1 gün veya 10GB’a ulaşınca yeni index’e geçiyor, 3 gün sonra warm tier’a iniyor ve 30 gün sonra siliniyor. Uyumluluk gerektiren ortamlarda silme yerine snapshot alıp cold storage’a gönderebilirsiniz.

Gerçek Dünya Senaryosu: DDoS Tespiti

Bir sabah alarmlar çalmaya başladı, sunucu yanıt veremez hale geldi. Kibana’yı açtım, son 15 dakikaya baktım. Discover ekranında tek bir IP’den 50.000 istek geldiğini gördüm. Şu sorgu ile tespit ettim:

Kibana KQL sorgusunda:

response_code: 200 and client_ip: * and @timestamp >= "now-15m"

Ardından Terms aggregation ile en çok istek atan IP’yi bulup anında Nginx’te engelledim:

sudo nano /etc/nginx/conf.d/blocklist.conf
geo $blocked_ip {
    default 0;
    1.2.3.4/32 1;
    10.0.0.0/8 0;
}

server {
    if ($blocked_ip) {
        return 444;
    }
}

ELK olmadan bu tespiti yapmam en az 30 dakika sürerdi. Kibana ile 2 dakika içinde kaynağı buldum.

Logstash Performans Ayarları

Yüksek trafik ortamlarında Logstash pipeline’ı darboğaz olabilir. /etc/logstash/logstash.yml dosyasında şu ayarları yapın:

pipeline.workers: 4
pipeline.batch.size: 500
pipeline.batch.delay: 50
queue.type: persisted
queue.max_bytes: 1gb
dead_letter_queue.enable: true
  • pipeline.workers: CPU çekirdek sayısı kadar veya bir eksik ayarlayın
  • pipeline.batch.size: Bir seferde işlenecek event sayısı, büyük değer daha iyi throughput ama daha fazla bellek demek
  • queue.type: persisted: Logstash çökerse bekleyen loglar kaybolmaz
  • dead_letter_queue.enable: Parse edilemeyen eventler buraya düşer, sessizce kaybolmaz

Sonuç

ELK Stack kurulumu ilk bakışta karmaşık görünüyor, özellikle Grok pattern’leri bazen sinir bozucu olabiliyor. Ama bir kere çalışır hale getirdiğinizde, log analizi bambaşka bir boyut kazanıyor. Artık “ne oldu” sorusunu yanıtlamak yerine “ne olabilir” sorusunu önceden yanıtlamaya başlıyorsunuz.

Önerdiğim yol haritası şöyle: Önce tek bir Nginx sunucusu ile başlayın, pipeline’ı oturuncaya kadar uğraşın. Sonra diğer sunucuları ekleyin. Dashboard’ları oluştururken “bu grafiği neden izliyorum, neye karar vermemi sağlıyor” sorusunu kendinize sorun. Her metrik bir aksiyona bağlı olmalı.

Production’a geçmeden önce şu konuları araştırmanızı öneririm: Elasticsearch güvenlik ayarları (TLS, rol tabanlı erişim), snapshot ve restore prosedürleri, Logstash dead letter queue yönetimi. Bu yazıda kasıtlı olarak bazı güvenlik ayarlarını basitleştirdim, prodüksiyonda mutlaka TLS’i etkinleştirin.

Log yönetimi bir araç meselesi değil, bir kültür meselesi. ELK Stack sizi daha iyi bir sistem yöneticisi yapmaz, ama doğru soruları sormayı kolaylaştırır.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir