Logstash Pipeline Tasarımı: Filter ve Grok Desenlerinde Uzmanlık

Production ortamında koşan yüzlerce mikroservisin loglarını merkezi bir elk stack’e toplamak işin sadece başlangıcıdır. Gerçek kaos, o anlamsız string yığınlarını anlamlı metrikler ve aranabilir alanlar haline getirmek istediğinizde, yani log parsing aşamasında başlar. Bu yazıda, logstash üzerinde performans canavarı bir pipeline nasıl tasarlanır, karmaşık log formatları grok desenleri ile CPU’yu ağlatmadan nasıl dize getirilir ve parsing hatalarıyla nasıl başa çıkılır derinlemesine inceleyeceğiz. Kahvenizi alın, çünkü regex (regular expressions) derinliklerine iniyoruz.

Grok ile Savaşmak: Regex Cehenneminden Nasıl Kurtuluruz?

Logstash dünyasında Grok, yapılandırılmamış (unstructured) metin verilerini yapılandırılmış (structured) JSON formatına dönüştürmek için kullandığımız en güçlü silahtır. Ancak yanlış ellerde bu silah, pipeline üzerinde ciddi darboğazlara (bottleneck) yol açabilir. Grok aslında arkada regex çalıştırır. Eğer yazdığınız regex deseni verimsiz ise, Logstash JVM heap limitlerini zorlar ve CPU tüketiminiz tavan yapar.

Farz edelim ki elimizde şöyle karmaşık, custom bir Java uygulama logu var:

2023-11-23T14:32:10.123Z [thread-pool-4] INFO  c.k.service.UserService [TX-98123] - User login failed for user 'kertenkerem_devops' from IP 192.168.1.50 - Reason: Invalid credentials

Bu logu parse etmek için yazacağımız klasik, hantal ve performansı düşük Grok filtresi genelde şuna benzer:

# TAVSİYE EDİLMEYEN YÖNTEM (Yavaş ve hantal)
filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:thread}\] %{LOGLEVEL:level}  %{NOTSPACE:class} \[%{DATA:tx_id}\] - %{GREEDYDATA:log_message}" }
  }
}

Neden tavsiye etmiyoruz? Çünkü %{DATA} ve %{GREEDYDATA} desenleri regex motorunu çok yorar. Logstash, eşleşme bulabilmek için metnin sonuna kadar gider, bulamazsa geri döner (backtracking). Bu da milyonlarca log satırında CPU fanlarınızın jet motoru gibi bağırmasına neden olur.

Doğru ve Optimize Grok Tasarımı

Aynı logu daha spesifik desenlerle, sınırları (anchors) belirleyerek parse edelim. Ayrıca kendi custom pattern’lerimizi tanımlayarak okunabilirliği artıralım:

filter {
  grok {
    pattern_definitions => {
      "THREAD_NAME" => "[a-zA-Z0-9-]+"
      "TX_ID" => "TX-[0-9]+"
    }
    match => { 
      "message" => "^%{TIMESTAMP_ISO8601:timestamp} \[%{THREAD_NAME:thread}\] %{LOGLEVEL:level}\s+%{NOTSPACE:class} \[%{TX_ID:tx_id}\] - (?<action>[a-zA-Z\s]+) '%{USER:username}' from IP %{IP:client_ip} - Reason: %{GREEDYDATA:reason}$" 
    }
  }
}

Neden bu daha iyi?

• Satır başı (^) ve satır sonu ($) işaretleyicilerini kullanarak regex motorunun tüm satırı gereksiz yere tekrar tekrar taramasını engelledik.
• %{DATA} yerine özel olarak tanımladığımız %{THREAD_NAME} ve regex limitli capture gruplarını kullandık.
• \s+ ile boşluk miktarı değişse bile eşleşmenin kırılmamasını sağladık.

Conditional Filters: Akıllı Pipeline Yönlendirmesi

Logstash pipeline’ınızda tüm log türlerini aynı Grok filtresinden geçirmek tam bir kaynak israfıdır. Nginx logu ile Java logu aynı pipeline’da karşılaşabilir. Çözüm, conditional (koşullu) filtreler kullanarak logları doğru parser’lara yönlendirmektir.

Aşağıdaki örnekte log kaynağına göre (örneğin Kubernetes namespace veya log tipi) nasıl filtreleme yapacağımızı görebilirsiniz:

filter {
  # Kubernetes metadata'sından gelen namespace bilgisine göre ayırma
  if [kubernetes][namespace] == "production" {
    mutate { add_tag => [ "prod_environment" ] }
    
    if [kubernetes][container_name] == "nginx-ingress" {
      grok {
        match => { "message" => "%{NGINXACCESS}" }
        remove_tag => [ "_grokparsefailure" ] # Başarılıysa default failure tag'ini ez
      }
    } else if [kubernetes][container_name] == "payment-api" {
      grok {
        match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
      }
    }
  } else {
    # Non-prod veya diğer loglar için hafif bir parser
    mutate { add_tag => [ "non_prod" ] }
  }
}

Bu yaklaşım sayesinde, sadece ilgili konteynerden gelen loglar ilgili Grok desenine maruz kalır. Boş yere tüm regex kuralları her log satırı için koşturulmaz.

Dead Letter Queue (DLQ) ve Parse Hatalarının Yönetimi

Günün birinde yazılımcı ekibinden biri log formatını değiştirir (çünkü neden değiştirmesinler ki?) ve pipeline patlar. Logstash bu durumda _grokparsefailure tag’ini basar ve logu Elasticsearch’e hatalı haliyle yazar. Eğer kritik bir audit logu parse edilemediği için kaybolursa başınız ağrıyabilir.

Logstash, teslim edilemeyen veya işlenemeyen event’ler için bir Dead Letter Queue (DLQ) mekanizması sunar. Bu mekanizmayı aktifleştirmek için logstash.yml dosyanızda şu ayarı yapmanız gerekir:

# logstash.yml
dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1024mb
path.dead_letter_queue: "/var/lib/logstash/dead_letter_queue"

DLQ’ya düşen bu bozuk logları okumak ve analiz etmek için ayrı bir Logstash pipeline’ı tasarlayabilirsiniz. Örneğin, hatalı logları tespit edip Slack’e alert atan veya bunları incelemek üzere ayrı bir “failed-logs” index’ine basan bir pipeline kurabiliriz:

# dlq_pipeline.conf
input {
  dead_letter_queue {
    path => "/var/lib/logstash/dead_letter_queue"
    commit_offsets => true
    pipeline_id => "main"
  }
}

output {
  elasticsearch {
    hosts => ["https://elasticsearch.kertenkerem.local:9200"]
    index => "logstash-dlq-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "highly_secure_password"
    ssl => true
    cacert => "/etc/logstash/certs/ca.crt"
  }
}

Logstash Performans Optimizasyonu: canavar gibi çalışan pipeline’lar

Eğer saniyede on binlerce log event’i işliyorsanız, Logstash tuning kaçınılmaz hale gelir. İşte production ortamlarında bizzat uyguladığımız hayat kurtaran performans tüyoları:

1. Grok Yerine Dissect Kullanın

Log formatınız belirli ve sabit bir ayırıcıya (delimiter) sahipse (örneğin Virgülle ayrılmış CSV, Tab veya Pipe | karakterleri), Grok kullanmayın! Dissect plugini, regex kullanmadan sadece string operasyonları ile parse işlemi yapar ve Grok’a göre neredeyse 10 kat daha hızlıdır.

# Dissect Örneği: Log formatı [Thread] LEVEL - Message şeklindeyse
filter {
  dissect {
    mapping => {
      "message" => "[%{thread}] %{level} - %{msg}"
    }
  }
}

2. Pipeline Workers Ayarı

Logstash varsayılan olarak CPU çekirdek sayınız kadar worker thread oluşturur. Ancak yoğun I/O işlemlerinde bu yetmeyebilir. logstash.yml içinde bu değeri optimize edin:

pipeline.workers: 8 # CPU core sayınızın 1.5 - 2 katı civarında tutabilirsiniz
pipeline.batch.size: 250 # Tek seferde Elasticsearch'e gönderilecek event sayısı

Batch size değerini artırmak throughput (iş çıkarma yeteneği) artışına sebep olurken, RAM tüketimini artırır. 125 ila 500 arasında ideal noktanızı test ederek bulmalısınız.

3. ECS (Elastic Common Schema) Uyumluluğu

Modern ELK mimarisinde log alanlarının (fields) standartlaştırılması çok önemlidir. Logstash filtrelerinizde yazdığınız output’ların ECS uyumlu olması, Kibana üzerinde dashboard yaparken veya SIEM kuralları yazarken hayatınızı kurtarır. Örneğin client_ip yerine source.ip, username yerine user.name kullanmaya özen gösterin.

Özet ve Kapanış

Verimli bir logstash pipeline tasarımı, sadece logları parse etmekle kalmaz; production ortamınızın gözlemleyebilirliğini (observability) artırırken altyapı maliyetlerinizi de düşürür. Grok yazarken regex sınırlarına dikkat etmek, conditional filter’lar ile akıllı routing yapmak ve DLQ ile hata yönetimini otomatikleştirmek sizi sıradan bir sistem yöneticisinden uzman bir DevOps mühendisine dönüştürür. Loglarınız temiz, kuyruklarınız boş olsun!