kertenkerem_
· DevOps · 4 dk okuma

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 yalnızca başlangıcıdır. Gerçek kaos, o anlamsız string yığınlarını anlamlı metriklere ve aranabilir alanlara dönüştürmek istediğinizde, yani log parsing aşamasında başlar. Bu yazıda Logstash üzerinde hızlı bir pipeline nasıl tasarlanır, karmaşık log formatları grok desenleriyle CPU’yu yormadan nasıl çözülür ve parsing hatalarıyla nasıl başa çıkılır, ayrıntılı inceleyeceğiz. Kahvenizi alın, regex (regular expressions) derinliklerine iniyoruz.

Grok ile savaşmak: regex cehenneminden nasıl kurtuluruz?

Logstash’te 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ü araçtır. Yanlış ellerde ise pipeline üzerinde ciddi darboğazlara (bottleneck) yol açabilir. Grok aslında arkada regex çalıştırır. Yazdığınız regex deseni verimsizse Logstash JVM heap limitlerini zorlar ve CPU tüketiminiz tavan yapar.

Elimizde şöyle karmaşık, custom bir Java uygulama logu olduğunu varsayalım:

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? %{DATA} ve %{GREEDYDATA} desenleri regex motorunu çok yorar. Logstash eşleşme bulabilmek için metnin sonuna kadar gider, bulamazsa geri döner (backtracking). Milyonlarca log satırında bu, 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. Kendi custom pattern’lerimizi de 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) filtrelerle 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şımla yalnızca ilgili konteynerden gelen loglar ilgili Grok desenine girer. Tüm regex kuralları her log satırı için boş yere 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 (neden değiştirmesinler ki?) ve pipeline patlar. Logstash bu durumda _grokparsefailure tag’ini basar ve logu Elasticsearch’e hatalı haliyle yazar. 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. Bunu etkinleş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 ya da 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: hızlı çalışan pipeline’lar

Saniyede on binlerce log event’i işliyorsanız Logstash tuning kaçınılmaz hale gelir. Production ortamlarında bizzat uyguladığımız performans tüyoları şunlar:

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 yalnızca string operasyonlarıyla 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. 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ışı sağlarken RAM tüketimini de 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 işinizi kolaylaştırı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ı logları parse etmenin ötesinde, production ortamınızın gözlemlenebilirliğini (observability) artırırken altyapı maliyetlerinizi de düşürür. Grok yazarken regex sınırlarına dikkat etmek, conditional filter’larla 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!