Multiline Log Ayrıştırma: Stack Trace ve Çok Satırlı Hata Loglarını İşleme

Bir Java uygulaması çöktüğünde, bir Python servisi beklenmedik bir hata fırlattığında ya da Node.js uygulamanız gece yarısı sessizce can çekişmeye başladığında, log dosyalarına bakarsınız. Ama çoğu zaman gördüğünüz şey, birbiriyle ilişkili onlarca satırdan oluşan bir kaos yığınıdır. Stack trace’ler, exception mesajları, iç içe geçmiş hata zincirleri… Bunları düzgün ayrıştıramadığınızda, bir olayın gerçekte ne zaman başladığını, kaç kez tekrarlandığını ve hangi bileşenden kaynaklandığını anlayamazsınız. Bu yazıda, çok satırlı log kayıtlarını nasıl işleyeceğinizi, stack trace’leri nasıl doğru şekilde ayrıştıracağınızı ve bu işi production ortamında güvenilir şekilde yapan araçları nasıl kuracağınızı ele alacağız.

Sorunun Özü: Neden Çok Satırlı Loglar Zor?

Standart log işleme araçlarının büyük çoğunluğu “her satır bir olay” mantığıyla çalışır. grep, awk, sed, hatta bazı log toplayıcıların varsayılan konfigürasyonları bu modeli benimser. Oysa gerçek dünyada bir Java NullPointerException 50 satır sürebilir, bir Python traceback 15-20 satır kaplabilir ve bir .NET exception zinciri bazen 100 satırı aşabilir.

Şu örnek bir Java stack trace’ine bakalım:

2024-01-15 14:32:11.445 ERROR [main] c.e.s.OrderService - Order processing failed
java.lang.NullPointerException: Cannot invoke method getPrice() on null object
    at com.example.service.OrderService.calculateTotal(OrderService.java:142)
    at com.example.service.OrderService.processOrder(OrderService.java:89)
    at com.example.controller.OrderController.submitOrder(OrderController.java:56)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.lang.reflect.Method.invoke(Method.java:498)
Caused by: com.example.exception.ProductNotFoundException: Product ID 9921 not found
    at com.example.repository.ProductRepository.findById(ProductRepository.java:67)
    ... 5 more

Bu bloğu tek bir olay olarak işleyemezseniz, log analiz sisteminiz bunu 10 ayrı log satırı olarak görür. Alerting sisteminiz yanlış sayar, log indekslemeniz anlamsız hale gelir ve arama yaparken ilgili satırları bir arada göremezsiniz.

Multiline Log Ayrıştırma Stratejileri

Çok satırlı logları işlemek için temelde iki yaklaşım vardır: başlangıç deseni ve bitiş deseni tabanlı gruplama.

Başlangıç deseni: Her yeni log kaydının belirli bir kalıpla başladığını varsayarsınız. Timestamp ile başlayan satırlar yeni bir kaydın başlangıcıdır, başlamayanlar bir önceki kaydın devamıdır.

Bitiş deseni: Belirli bir kalıp görüldüğünde log bloğunun bittiğini kabul edersiniz. Java’da } ile biten satırlar veya belirli marker kelimeler buna örnek verilebilir.

Pratikte başlangıç deseni çok daha yaygın ve güvenilirdir.

Filebeat ile Multiline Konfigürasyonu

Filebeat, ELK Stack’in log toplayıcısıdır ve multiline desteği oldukça gelişmiştir. Production ortamında en çok kullanılan konfigürasyon şöyle görünür:

# /etc/filebeat/filebeat.yml
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/myapp/*.log
    multiline.type: pattern
    multiline.pattern: '^d{4}-d{2}-d{2}'
    multiline.negate: true
    multiline.match: after
    multiline.max_lines: 500
    multiline.timeout: 5s
    fields:
      app_name: order-service
      environment: production

Burada kritik parametreler şunlardır:

• multiline.pattern: Yeni bir log kaydının başladığını gösteren regex. Tarih formatına uyan satırlar yeni kayıt başlangıcıdır.
• multiline.negate: true: Pattern’e uymayan satırları bir öncekiyle birleştir demektir.
• multiline.match: after: Pattern’e uyan satırdan sonra gelen eşleşmeyenleri bu kayda ekle.
• multiline.max_lines: Tek bir kayıtta maksimum satır sayısı. Sonsuz döngüyü önler.
• multiline.timeout: Bu süre geçince beklemeden gönder.

Python uygulaması logları için pattern biraz farklı olabilir:

# Python traceback için konfigürasyon
filebeat.inputs:
  - type: log
    paths:
      - /var/log/pythonapp/app.log
    multiline.type: pattern
    multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'
    multiline.negate: true
    multiline.match: after
    multiline.max_lines: 200

Fluentd ile Gelişmiş Multiline İşleme

Fluentd, özellikle Kubernetes ortamlarında yaygın kullanılan bir log toplayıcıdır ve multiline parsing için çok daha esnek bir yapı sunar.

# /etc/td-agent/td-agent.conf
<source>
  @type tail
  path /var/log/app/*.log
  pos_file /var/log/td-agent/app.log.pos
  tag app.logs
  
  <parse>
    @type multiline
    format_firstline /^d{4}-d{2}-d{2} d{2}:d{2}:d{2}/
    format1 /^(?<time>d{4}-d{2}-d{2} d{2}:d{2}:d{2}.d{3}) (?<level>[A-Z]+) [(?<thread>[^]]+)] (?<logger>[^ ]+) - (?<message>.*)/
    multiline_flush_interval 5s
  </parse>
</source>

<filter app.logs>
  @type grep
  <regexp>
    key level
    pattern /ERROR|WARN|FATAL/
  </regexp>
</filter>

<match app.logs>
  @type elasticsearch
  host localhost
  port 9200
  index_name app-logs
  type_name _doc
</match>

Fluentd’nin güçlü yanı, format_firstline ile yeni kaydı tespit ettikten sonra format1, format2 gibi ek formatlarla kaydın içindeki alanları da parse edebilmenizdir. Bu sayede hem birleştirme hem yapılandırılmış ayrıştırma tek adımda gerçekleşir.

Logstash Grok Patterns ile Stack Trace İşleme

Logstash, özellikle karmaşık log formatlarını işlemek için Grok pattern’larını kullanır. Stack trace içeren loglar için şu konfigürasyonu kullanabilirsiniz:

# /etc/logstash/conf.d/java-app.conf
input {
  beats {
    port => 5044
  }
}

filter {
  if [fields][app_name] == "order-service" {
    grok {
      match => {
        "message" => [
          "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:log_level} [%{DATA:thread}] %{DATA:logger} - %{GREEDYDATA:error_message}",
          "%{GREEDYDATA:raw_message}"
        ]
      }
    }
    
    # Stack trace içerip içermediğini kontrol et
    if [error_message] =~ /Exception|Error|Caused by/ {
      mutate {
        add_tag => ["has_stacktrace"]
      }
      
      # Exception tipini çıkar
      grok {
        match => {
          "error_message" => "(?<exception_type>[a-zA-Z.]+Exception|[a-zA-Z.]+Error)"
        }
        tag_on_failure => []
      }
    }
    
    date {
      match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
      target => "@timestamp"
    }
  }
}

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

Bu konfigürasyonla Logstash hem multiline log bloklarını tek kayıt olarak alır (Filebeat zaten birleştirmiştir), hem de içindeki alanları ayrıştırır, hem de stack trace varlığını etiketler.

Python ile Custom Log Parser Yazmak

Bazen hazır araçlar ihtiyacınızı karşılamaz ve kendiniz bir parser yazmanız gerekir. Özellikle standart dışı log formatlarında veya log verilerini başka sistemlere aktarırken bu durum ortaya çıkar.

#!/usr/bin/env python3
# multiline_parser.py

import re
import json
from datetime import datetime
from typing import Generator, Dict, Any

class MultilineLogParser:
    def __init__(self, log_file: str):
        self.log_file = log_file
        # Yeni log kaydının başlangıcını belirleyen pattern
        self.entry_pattern = re.compile(
            r'^(d{4}-d{2}-d{2} d{2}:d{2}:d{2}[.,]d{3})'
        )
        self.exception_pattern = re.compile(
            r'(?P<exc_type>[w.]+(?:Exception|Error)):s*(?P<exc_msg>.*)'
        )
        self.caused_by_pattern = re.compile(
            r'Caused by:s+(?P<exc_type>[w.]+(?:Exception|Error)):s*(?P<exc_msg>.*)'
        )
    
    def parse_entries(self) -> Generator[Dict[str, Any], None, None]:
        current_lines = []
        
        with open(self.log_file, 'r', encoding='utf-8', errors='replace') as f:
            for line in f:
                line = line.rstrip('n')
                
                if self.entry_pattern.match(line) and current_lines:
                    yield self._process_entry(current_lines)
                    current_lines = [line]
                else:
                    current_lines.append(line)
        
        if current_lines:
            yield self._process_entry(current_lines)
    
    def _process_entry(self, lines: list) -> Dict[str, Any]:
        full_message = 'n'.join(lines)
        entry = {
            'raw': full_message,
            'line_count': len(lines),
            'has_stacktrace': False,
            'exceptions': [],
            'stack_frames': []
        }
        
        # Stack trace var mı?
        if len(lines) > 1:
            stack_lines = [l for l in lines if l.strip().startswith('at ')]
            if stack_lines:
                entry['has_stacktrace'] = True
                entry['stack_depth'] = len(stack_lines)
                entry['stack_frames'] = [
                    l.strip().replace('at ', '') for l in stack_lines[:10]
                ]
        
        # Exception tiplerini çıkar
        for line in lines:
            exc_match = self.exception_pattern.search(line)
            if exc_match:
                entry['exceptions'].append({
                    'type': exc_match.group('exc_type'),
                    'message': exc_match.group('exc_msg'),
                    'caused_by': 'Caused by' in line
                })
        
        return entry

if __name__ == '__main__':
    parser = MultilineLogParser('/var/log/myapp/app.log')
    for entry in parser.parse_entries():
        if entry['has_stacktrace']:
            print(json.dumps(entry, ensure_ascii=False, indent=2))

Bu parser’ı cron job ile çalıştırıp çıktıyı bir veritabanına veya alerting sistemine besleyebilirsiniz.

Kubernetes Ortamında Multiline Log Zorlukları

Kubernetes’te her container kendi stdout/stderr’ine yazar ve container runtime bunu satır satır dosyaya kaydeder. Bu durum, bir Java uygulamasının 50 satırlık stack trace’ini 50 ayrı satır olarak görmek anlamına gelir.

Kubernetes için özel bir Filebeat DaemonSet konfigürasyonu:

# filebeat-daemonset-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: filebeat-config
  namespace: kube-system
data:
  filebeat.yml: |
    filebeat.autodiscover:
      providers:
        - type: kubernetes
          node: ${NODE_NAME}
          hints.enabled: true
          hints.default_config:
            type: container
            paths:
              - /var/log/containers/*${data.kubernetes.container.id}.log
            multiline.pattern: '^{'
            multiline.negate: true
            multiline.match: after
    
    processors:
      - add_kubernetes_metadata:
          host: ${NODE_NAME}
          matchers:
            - logs_path:
                logs_path: "/var/log/containers/"
      - decode_json_fields:
          fields: ["message"]
          target: ""
          overwrite_keys: true
    
    output.elasticsearch:
      hosts: ['${ELASTICSEARCH_HOST:elasticsearch}:9200']
      index: "k8s-logs-%{+yyyy.MM.dd}"

Kubernetes annotation’larıyla da per-pod multiline konfigürasyonu yapabilirsiniz:

# Pod tanımına eklenecek annotation'lar
annotations:
  co.elastic.logs/multiline.pattern: '^d{4}-d{2}-d{2}'
  co.elastic.logs/multiline.negate: "true"
  co.elastic.logs/multiline.match: "after"
  co.elastic.logs/multiline.max_lines: "300"

Bash ile Hızlı Stack Trace Analizi

Bazen production’da acil bir durum var ve karmaşık araç kurma fırsatınız yoktur. Saf bash ile stack trace’leri analiz edebilirsiniz:

#!/bin/bash
# stacktrace_analyzer.sh - Hizli stack trace analizi

LOG_FILE="${1:-/var/log/app/app.log}"
OUTPUT_DIR="/tmp/stacktrace_analysis_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "Log dosyasi analiz ediliyor: $LOG_FILE"

# Exception tiplerini say ve sirala
echo "=== En Sik Gorulen Exception Tipleri ==="
grep -oP '[w.]+(?:Exception|Error)(?=:)' "$LOG_FILE" | 
  sort | uniq -c | sort -rn | head -20 | 
  tee "$OUTPUT_DIR/exception_counts.txt"

# Stack trace bloklarini ayristir
echo ""
echo "=== Stack Trace Bloklari Cikartiliyor ==="
awk '
/[0-9]{4}-[0-9]{2}-[0-9]{2}.*ERROR/ {
  if (block != "" && in_stacktrace) {
    print "---BLOCK_END---"
    print block
    block_count++
  }
  in_stacktrace = 0
  block = $0
  next
}
/^s+at / {
  in_stacktrace = 1
  block = block "n" $0
  next
}
/^(Caused by:|java.|com.|org.)/ {
  if (in_stacktrace) {
    block = block "n" $0
  }
  next
}
{
  if (!in_stacktrace) {
    block = $0
  }
}
END {
  print "Toplam stack trace bloku: " block_count
}
' "$LOG_FILE" > "$OUTPUT_DIR/stacktraces.txt"

# Saatlik hata dagilimi
echo ""
echo "=== Saatlik Hata Dagilimi ==="
grep "ERROR|FATAL" "$LOG_FILE" | 
  grep -oP '^d{4}-d{2}-d{2} d{2}' | 
  sort | uniq -c | 
  awk '{printf "  %s:00 - %d hatan", $2, $1}' | 
  tee "$OUTPUT_DIR/hourly_errors.txt"

echo ""
echo "Analiz tamamlandi. Sonuclar: $OUTPUT_DIR"

Alertmanager ile Stack Trace Tabanlı Uyarılar

Stack trace’leri doğru ayrıştırdıktan sonra, bu bilgiyi alerting sistemine entegre etmek kritik önem taşır. Prometheus + Alertmanager kombinasyonunda Elasticsearch’ten gelen log metriklerini kullanabilirsiniz:

# alerting-rules.yml
groups:
  - name: application-errors
    interval: 1m
    rules:
      - alert: HighStackTraceRate
        expr: |
          sum(rate(elasticsearch_log_entries_total{
            has_stacktrace="true",
            environment="production"
          }[5m])) by (app_name, exception_type) > 0.5
        for: 2m
        labels:
          severity: critical
          team: backend
        annotations:
          summary: "{{ $labels.app_name }} yuksek stack trace orani"
          description: |
            {{ $labels.app_name }} uygulamasinda son 5 dakikada
            {{ $labels.exception_type }} tipi exception hizi
            dakikada 30'u gecti. Deger: {{ $value | humanize }}
          runbook_url: "https://wiki.company.com/runbooks/{{ $labels.exception_type }}"
      
      - alert: NewExceptionType
        expr: |
          count by (app_name, exception_type) (
            elasticsearch_log_entries_total{has_stacktrace="true"}
          ) unless on(app_name, exception_type)
          (elasticsearch_log_entries_total{has_stacktrace="true"} offset 1d)
        for: 0m
        labels:
          severity: warning
        annotations:
          summary: "{{ $labels.app_name }}: Yeni exception tipi goruldu"
          description: "{{ $labels.exception_type }} daha once hic gorulmemisti"

Gerçek Dünya Senaryosu: E-ticaret Platformunda Kriz Yönetimi

Birkaç yıl önce karşılaştığım bir olayı paylaşayım. Büyük bir e-ticaret platformunun ödeme servisi gece 02:30’da yavaşlamaya başladı. Log dosyaları incelendiğinde her şey normal görünüyordu, çünkü monitoring sistemi “her satır bir olay” modeliyle çalışıyordu ve stack trace’lerin ilk satırı genellikle INFO seviyesindeydi.

Sorunu çözmek için şu adımları izledik. Önce ham log dosyasına bakıldı ve gerçekte her “başarılı” transaction log satırının arkasında 3-4 satırlık suppressed exception olduğu görüldü. Sonra multiline parser devreye alındı ve aslında dakikada 400+ ConnectionPoolTimeoutException fırlatıldığı ortaya çıktı. Log analizi doğru yapıldığında, sorunun veritabanı bağlantı havuzunda tükenme olduğu 10 dakikada tespit edildi. Düzeltme ise bağlantı havuzu boyutunu artırıp idle timeout değerini düşürmekten ibaret kaldı.

Eğer multiline log ayrıştırma düzgün kurulu olsaydı, bu sorunu gece yarısı değil akşam saatlerinde, ilk belirtiler ortaya çıktığında yakalayabilirdik.

Performans Optimizasyonu: Büyük Log Dosyalarında Dikkat Edilecekler

Multiline log işleme, tek satır işlemeden doğal olarak daha maliyetlidir. Production ortamında şu noktalara dikkat edin:

• Buffer boyutlarını doğru ayarlayın: Filebeat’te max_lines değerini gereğinden büyük tutmayın. 500 satır genellikle yeterlidir, 5000 değil.
• Timeout değerlerini optimize edin: multiline.timeout: 5s çoğu senaryo için uygundur. Gerçek zamanlı sistemlerde 2s daha iyidir.
• Regex karmaşıklığını minimize edin: Timestamp pattern’ı için ^d{4}-d{2}-d{2} basit bir regex kullanın, gereksiz capture group’lardan kaçının.
• Paralel işleme: Fluentd’de direktifini kullanarak yük dengelemesi yapın. Tek bir büyük log dosyası için worker sayısını artırmak her zaman işe yaramaz ama farklı uygulamalardan gelen logları paralel işlemek belirgin fark yaratır.
• Log rotation ile uyumu test edin: Logrotate devreye girdiğinde, o an işlenen multiline bloğun kaybolmaması için copytruncate yerine create modunu tercih edin ve toplayıcıyı yeniden okuma tetikleyecek şekilde konfigüre edin.

Sonuç

Multiline log ayrıştırma, birçok ekibin hafife aldığı ama production olaylarında büyük fark yaratan bir konudur. Stack trace’leri doğru şekilde tek bir olay olarak gruplayamazsanız, hata sayılarınız yanlış olur, alertler hem gürültülü hem de eksik kalır ve gerçek sorunun izini sürmek zorlaşır.

Bu yazıda ele aldığımız yaklaşımları özetlemek gerekirse: Filebeat ve Fluentd gibi modern log toplayıcıların multiline yeteneklerini aktif olarak kullanın, Kubernetes ortamında annotation bazlı konfigürasyondan yararlanın, acil durumlarda bash araçlarıyla hızlı analiz yapabilir olun ve stack trace bilgisini alerting sistemine taşıyarak proaktif uyarılar kurun.

En önemli tavsiye şudur: Bu konfigürasyonları production’a geçmeden önce gerçek stack trace örnekleriyle test edin. Her uygulamanın log formatı biraz farklıdır ve varsayılan pattern’lar her zaman işe yaramaz. Bir saatlik doğru konfigürasyon çalışması, gece yarısı saatlerce süren kriz yönetiminin önüne geçebilir.