Yapılandırılmış loglama ve korelasyon kimliği: hata ayıklanabilir otomasyon

Yapılandırılmış loglama, bir otomasyonun gece üçte neden durduğunu sabah kahve soğumadan anlayabilmenin en ucuz yolu. Yıllarca serbest metin log yazdım: print ile başlayan, sonra logging.info("dosya işlendi: %s", ad) gibi satırlara evrilen, ama sonunda bir metin editöründe Ctrl+F ile aranan dosyalar. Üç dört adımlı, günde bir kez koşan bir betikte bu yeter. Sorun, aynı gün içinde birkaç kez tetiklenen, alt süreçlere dallanan ve paralel çalışan işlerde başlıyor: hangi satırın hangi koşuma ait olduğu artık kestirilemiyor.

Bu yazıda kendi otomasyonlarımda oturttuğum düzeni anlatacağım: her log satırı bir JSON nesnesi, her koşumun bir korelasyon kimliği (run_id) var ve bu kimlik elle taşınmadan tüm alt adımlara, yardımcı fonksiyonlara ve iş parçacıklarına ulaşıyor. Python'un standart logging modülü ve contextvars ile bunu ek bağımlılık olmadan kurmak mümkün. Ardından seviye stratejisine, sır sızdırmayan maskelemeye ve dönen dosya yönetimine geçeceğim; sonunda da jq ile tek koşumu birkaç saniyede ayıklamayı göstereceğim.

Örnekler Python 3.10 ve üzeri için yazıldı ve bu yazı için baştan sona çalıştırıldı; jq komutları Linux ve macOS'ta doğrudan, Windows'ta Git Bash ya da WSL altında aynı şekilde çalışır.

Serbest metin log neden yetmiyor

Serbest metin logun sorunu yazarken değil okurken ortaya çıkıyor. Metin satırı insan için tasarlanmıştır; makine için bir desen tanıma problemidir. grep ile başlarsınız, sonra bir düzenli ifade yazarsınız, sonra biri mesaj metnini değiştirir ve o ifade sessizce boş döner. Bu döngüyü yeterince yaşadıktan sonra dört sorunu netleştirdim:

• Alan sınırı yok. Tarih, seviye ve mesaj boşlukla ayrılır; mesajın içinde boşluk ve iki nokta olduğu anda ayrıştırma bozulur.
• Bağlam kaybolur. "Bağlantı hatası" satırı hangi dosya, hangi adım, hangi koşum için yazıldı? Yazan bilir, okuyan bilmez.
• Paralel koşumlar iç içe geçer. İki işin satırları aynı dosyada birbirine karışır; zaman damgası sıralamaya yeter, ayırmaya yetmez.
• Araç desteği yok. jq, DuckDB ya da Grafana Loki gibi araçlar JSON satırı doğrudan yer; serbest metin için önce bir ayrıştırıcı yazmanız gerekir.

JSON satır biçimi (JSON Lines ya da NDJSON) bu dört sorunu tek hamlede çözüyor: her satır bağımsız bir nesne, her nesne aynı alanlara sahip ve mesajın içinde ne olursa olsun ayrıştırma bozulmuyor. Bedeli, dosyanın çıplak gözle biraz daha zor okunması; jq bunu fazlasıyla telafi ediyor.

run_id ve contextvars ile otomatik alan enjeksiyonu

Korelasyon kimliği fikri web dünyasından geliyor: her HTTP isteğine bir kimlik verilir ve istek hangi servisten geçerse geçsin bu kimlik başlıklarda taşınır. Toplu işler için aynı fikir daha da basit: koşum başladığında bir run_id üretilir, koşum bitene kadar her satıra yazılır. Zor kısım, bu kimliği her log çağrısına elle vermeden taşımak. Fonksiyon imzalarına run_id parametresi eklemeye başlarsanız kod kısa sürede çirkinleşir; global değişken kullanırsanız paralel koşumlar birbirinin kimliğini ezer.

contextvars tam bu boşluğu dolduruyor. Bir ContextVar, çalıştığı bağlama özel bir değer taşır; log filtresi de kaydı oluştururken bu değeri okuyup LogRecord üzerine ekler. Biçimleyici, filtre ve küçük bir adım yöneticisi şöyle:

import json
import logging
import uuid
from contextlib import contextmanager
from contextvars import ContextVar
from datetime import datetime, timezone

run_id_var: ContextVar[str] = ContextVar("run_id", default="-")
step_var: ContextVar[str] = ContextVar("step", default="-")


class BaglamFiltresi(logging.Filter):
    """Her kayda run_id ve step alanlarını otomatik ekler."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.run_id = run_id_var.get()
        record.step = step_var.get()
        return True


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        zaman = datetime.fromtimestamp(record.created, tz=timezone.utc)
        satir = {
            "ts": zaman.isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "run_id": getattr(record, "run_id", "-"),
            "step": getattr(record, "step", "-"),
            "msg": record.getMessage(),
        }
        ekstra = getattr(record, "alan", None)   # log.info("...", extra={"alan": {...}})
        if ekstra:
            satir.update(ekstra)
        if record.exc_info:
            satir["exc"] = self.formatException(record.exc_info)
        return json.dumps(satir, ensure_ascii=False)


def logger_kur(ad: str = "otomasyon") -> logging.Logger:
    logger = logging.getLogger(ad)
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    handler.addFilter(BaglamFiltresi())
    logger.addHandler(handler)
    return logger


log = logger_kur()


def kosum_baslat() -> str:
    run_id = uuid.uuid4().hex[:12]
    run_id_var.set(run_id)
    return run_id


@contextmanager
def adim(ad: str):
    token = step_var.set(ad)
    log.info("adım başladı")
    try:
        yield
    finally:
        step_var.reset(token)


if __name__ == "__main__":
    kosum_baslat()
    with adim("indir"):
        log.info("dosyalar indirildi", extra={"alan": {"dosya": 3}})
    with adim("isle"):
        try:
            1 / 0
        except ZeroDivisionError:
            log.exception("hesap adımı çöktü")
    log.info("koşum bitti", extra={"alan": {"durum": "hata", "sure_sn": 1.8}})

Çıktı, her satırda run_id ve step alanları dolu, hata satırında ise exc alanında tam izleme bulunan JSON satırları. Birkaç noktaya dikkat:

• record.created kullanmak, zaman damgasının log çağrısı anını yansıtmasını sağlar; biçimleyicinin içinde datetime.now() çağırmak kuyruklu handler'larda yanıltır.
• Filtre handler'a takılı. Böylece alt logger'lardan yayılan (propagate) kayıtlar da aynı alanları alır; logger'a takılan filtre yalnız o logger'ın kendi kayıtlarını görür.
• extra={"alan": {...}} ile verilen sözlük doğrudan satıra açılır; sayılar, süreler ve durum bilgisi mesaj metnine gömülmek yerine ayrı alan olur.
• asyncio görevleri bağlamı otomatik kopyalar. ThreadPoolExecutor ile iş dağıtıyorsanız görevi contextvars.copy_context().run(fn, ...) ile sarın; yoksa iş parçacığında run_id varsayılan değere düşer.
• adim bağlam yöneticisi bittiğinde reset ile değeri geri alır; iç içe adımlarda üst adımın adı kaybolmaz.

Seviye stratejisi: neyi INFO, neyi ERROR yazmalı

Seviyelerin en sık gördüğüm kötüye kullanımı her şeyin INFO olması; ikincisi, yakalanan her istisnanın ERROR olarak yazılıp aynı hatanın üç katmanda üç kez loglanması. İkisi de logu gürültüye boğar ve asıl uyarıyı görünmez kılar. Benim uyguladığım sözleşme şu:

• DEBUG: kayıt başına ayrıntı; satır numarası, ham cevabın boyutu, denenen URL. Konsolda açık, dosyada gerektiğinde açılır.
• INFO: koşum ve adım sınırları, işlenen kayıt sayıları, özet süreler. Bir koşumun INFO satırlarını okuyan kişi ne olduğunu anlamalı ama boğulmamalı.
• WARNING: iş devam etti ama bir şey beklenenden farklıydı: yeniden deneme sonrası başarı, boş gelen dosya, eksik ama zorunlu olmayan alan. Sabah toplu bakılacak şeyler.
• ERROR: koşum ya da adım hedefine ulaşamadı ve bir insanın bakması gerekiyor. Her ERROR satırı bir bildirim tetikleyebilmeli; tetiklemeyecekse ERROR değildir.
• CRITICAL: yalnız süreç devam edemiyorsa. Yılda birkaç kez görmüyorsanız muhtemelen hiç kullanmıyorsunuz, bu da normaldir.

İki tamamlayıcı kural var. İstisna en üst katmanda bir kez, log.exception(...) ile ve izleme dahil yazılır; alt katmanlar yakalayıp yeniden fırlatır, loglamaz. İkincisi, her koşum tek bir bitiş satırıyla kapanır: durum, süre, sayılar. O satır yoksa koşum yarım kalmış demektir; nöbetçi betiklerim tam olarak bu satırın yokluğunu arar.

Bu sözleşme seviye alanını bir bildirim sistemine bağlamayı da mümkün kılıyor. Örneğin PowerShell ile sunucu izleme betiğimde eşik aşımını WARNING, toplama hatasını ERROR olarak yazıyorum; bildirimi tetikleyen tek şey seviye alanı, mesaj metnini hiç okumuyor.

Sır maskeleme ve dönen dosya boyutu

Log dosyası genellikle sistemin en az korunan parçasıdır: yedeklere girer, destek için paylaşılır, bazen bir sohbet penceresine yapıştırılır. İçinde bir API anahtarı ya da Bearer token varsa o sır artık her yerdedir. Birinci savunma sırrı hiç loglamamak: URL'leri sorgu parametresiz, başlıkları yalnız adıyla yazın. İkinci savunma, bu kuralı bir kez unutmanızı affeden bir maskeleme filtresi. Aynı yerde dönen dosya yapılandırması da duruyor; sonsuz büyüyen bir log dosyası bir gün diski doldurur ve o gün genellikle başka bir şeyin de bozulduğu gündür.

import logging
import re
from logging.handlers import RotatingFileHandler

from kod1_jsonlog import BaglamFiltresi, JsonFormatter, log, kosum_baslat, adim

GIZLI_DESENLER = [
    (re.compile(r"(?i)\b(api[_-]?key|token|password|parola|secret)\s*[=:]\s*\S+"),
     r"\1=[MASKE]"),
    (re.compile(r"Bearer\s+[A-Za-z0-9._\-]+"), "Bearer [MASKE]"),
]


def maskele(metin: str) -> str:
    for desen, yerine in GIZLI_DESENLER:
        metin = desen.sub(yerine, metin)
    return metin


class MaskeFiltresi(logging.Filter):
    """Mesajı biçimlendirir ve maskeler; sonraki handler'lar maskeli metni görür."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.msg = maskele(record.getMessage())
        record.args = None
        return True


class MaskeliJsonFormatter(JsonFormatter):
    def formatException(self, ei) -> str:
        return maskele(super().formatException(ei))


def dosya_handler(yol: str) -> RotatingFileHandler:
    h = RotatingFileHandler(
        yol, maxBytes=20 * 1024 * 1024, backupCount=5, encoding="utf-8"
    )
    h.setLevel(logging.INFO)              # DEBUG konsolda kalır, dosyaya girmez
    h.addFilter(BaglamFiltresi())
    h.setFormatter(MaskeliJsonFormatter())
    return h


log.addHandler(dosya_handler("otomasyon.log"))
for h in log.handlers:                    # konsol dahil her handler maskeler
    h.addFilter(MaskeFiltresi())

if __name__ == "__main__":
    kosum_baslat()
    with adim("giris"):
        log.info("oturum açıldı, api_key=sk-ornek-123456 kullanici=servis")
        log.debug("ham başlık: Authorization: Bearer eyJhbGciOi.ornek.imza")
    log.info("koşum bitti", extra={"alan": {"durum": "ok"}})

Bu kodu çalıştırdığımda konsolda ve dosyada api_key=[MASKE] ile Bearer [MASKE] görünüyor; ham değerler hiçbir çıktıya ulaşmıyor. Tasarım kararları:

• Filtre getMessage() ile mesajı biçimlendirir, maskeler ve args alanını temizler. Aynı LogRecord nesnesi tüm handler'lara gittiği için ilk handler maskeledikten sonra diğerleri de maskeli metni görür; yine de filtreyi her handler'a takmak, handler sırasına bağımlılığı ortadan kaldırır.
• Desenler bilinçli olarak geniş. Yanlış pozitif bir maskeleme, sızan bir token'dan çok daha ucuzdur.
• İstisna izleri ayrı bir kanaldan geçtiği için formatException da maskeden geçirilir; aksi hâlde bir hata mesajındaki URL parametresi sızabilir.
• RotatingFileHandler boyuta göre döner: 20 MB ve beş yedek ile dosya en fazla 120 MB yer kaplar. Zaman bazlı dönüş için TimedRotatingFileHandler var; ben boyut bazlıyı tercih ediyorum, çünkü diski koruyan şey tarih değil boyuttur.
• Windows'ta dönüş anında dosyayı başka bir süreç açık tutuyorsa yeniden adlandırma başarısız olur. Log dosyasına tek bir yazar süreç bırakın; çok süreçli işlerde her sürece kendi dosyasını verin, run_id zaten hepsini birleştirir.

jq ile tek koşumu ayıklamak

Tüm bu yapının karşılığı burada alınıyor. Log artık bir veri kümesi, jq da onun sorgu dili. jq bir dosyadaki ardışık JSON nesnelerini doğrudan akış olarak işler; satır başına bir nesne yazdığınız sürece ek bir hazırlık gerekmez.

# 1) Tek koşumun tüm satırları, zaman sırasıyla
jq -c 'select(.run_id == "3f9a1c2b7d4e")' otomasyon.log

# 2) Aynı koşumun yalnız hataları: zaman, adım, mesaj
jq -r 'select(.run_id == "3f9a1c2b7d4e" and .level == "ERROR")
       | "\(.ts) \(.step): \(.msg)"' otomasyon.log

# 3) Son koşumun kimliğini al, adım başına satır say
RUN=$(tail -n 1 otomasyon.log | jq -r '.run_id')
jq -r --arg r "$RUN" 'select(.run_id == $r) | .step' otomasyon.log | sort | uniq -c

# 4) Hangi koşumlar hata verdi? Kimlik ve hata sayısı
jq -r 'select(.level == "ERROR") | .run_id' otomasyon.log | sort | uniq -c | sort -rn

• İlk komut bir koşumun tüm satırlarını sırayla verir; -c her nesneyi tek satırda basar.
• İkincisi yalnız hataları, adım adıyla birlikte okunur biçimde çıkarır. Gece üç sorusunun cevabı çoğu zaman bu iki komutta biter.
• Üçüncüsü son koşumun kimliğini alır ve adım başına satır sayar. Bir adımın beklenenden az satır üretmesi, hata vermeden sessizce atlandığının ilk işaretidir.
• Dördüncüsü hangi koşumların hata verdiğini sayar; haftalık bakışta hangi günlerin sorunlu olduğunu tek komutla gösterir.

Aynı run_id değerini işin durumunu tuttuğum tabloya da yazıyorum. Böylece SQLite ile küçük veri boru hatları yazısında anlattığım koşum tablosu ile log dosyası tek bir anahtar üzerinden birleşiyor. Log "ne oldu" sorusunu, tablo "sonuç ne" sorusunu cevaplar; ikisini birleştiren şey kimliktir.

Bu düzenin kurulumu bir öğleden sonra sürüyor: bir biçimleyici, iki filtre, bir handler ve bir sözleşme. Getirisi ise her arızada hissediliyor; dosyayı açıp kaydırmak yerine kimliği sorup cevabı almak. Serbest metin logdan JSON satıra geçmek için mevcut logging.info çağrılarını değiştirmeniz de gerekmiyor, yalnız handler değişiyor. Küçük ama dayanıklı otomasyonlar kurmak veri ve otomasyon danışmanlığı işimin temel yaklaşımı. Elinizdeki betikleri hata ayıklanabilir hâle getirmek isterseniz iletişim sayfasından bana yazabilirsiniz; hangi kimliğin nereye taşınacağını ve hangi satırın bildirim tetikleyeceğini birlikte netleştiririz.