Yüksek eşzamanlılıkta Python uygulamaları dünyasında, kararlı bir hizmet ile tam bir kesinti arasındaki fark genellikle iş mantığında değil, sistemin hatayı nasıl gözlemlediği ve buna nasıl tepki verdiğiyle ilgilidir. Geleneksel print ifadeleri ve temel logging.basicConfig çağrıları, uygulamanız binlerce eşzamanlı isteği işlediğinde yetersiz kalır. Gecikme arttığında veya hatalar oluştuğunda, geliştiricilerin sorunları gerçek zamanlı olarak teşhis edebilmeleri için kesin, makine tarafından okunabilir verilere ihtiyaçları vardır. Bu yazı, asenkron Python ortamlarına özel sağlam, yapılandırılmış günlük kaydı ve taneli hata yönetimi desenlerinin nasıl uygulanacağını incelemektedir.
Yapılandırılmış Günlük Kayıt İçin Zorunluluk
Yapılandırılmış günlük kayıt, günlük mesajlarını düz metin yerine yapılandırılmış veri olarak, genellikle JSON formatında biçimlendirme pratiğidir. Bu yaklaşım, dağıtık sistemlerde kritiktir çünkü log birleştirme araçlarının (ELK Stack, Datadog veya Splunk gibi) log kayıtlarını verimli bir şekilde ayrıştırmasına, filtrelemesine ve indekslemesine olanak tanır. Eşzamanlı bir ortamda, bir istek kimliği, kullanıcı kimliği, gecikme ve durum kodu içeren tek bir günlük satırı, belirli bir işlemin mikro hizmetler arasında anında izlenmesini sağlar. Bunu Python'da gerçekleştirmek için standart"Yapılandırılmış veri olmadan günlük kayıtlar sadece gürültüdür. Bağlam olmadan hatalar sadece tahmindir."
logging modülüyle entegre olan python-json-logger veya benzeri kütüphanelerden yararlanmalısınız. Hedef, her günlük kaydının gerçeklerin kendi kendine yeterli bir sözlük olmasını sağlamaktır.
Uygulama Örneği
Bağlamı otomatik olarak JSON'a dönüştüren üretim hazır bir günlükçü nasıl yapılandırılır, işte böyle:import logging
import json
from pythonjsonlogger import jsonlogger
import sys
def setup_structured_logger(level=logging.INFO):
logger = logging.getLogger("app")
logger.setLevel(level)
# stderr'e yazan (veya üretimde bir dosyaya) bir işleyici oluşturun
handler = logging.StreamHandler(sys.stdout)
# Özel bir JSON biçimlendirici ayarlayın
formatter = jsonlogger.JsonFormatter(
fmt='%(asctime)s %(name)s %(levelname)s %(message)s %(process)d',
datefmt='%Y-%m-%d %H:%M:%S',
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
logger = setup_structured_logger()
def process_request(request_id, user_id, data):
logger.info(
"İşlem yapılıyor",
extra={
"request_id": request_id,
"user_id": user_id,
"action": "validate_data",
"payload_size": len(data)
}
)
# İş mantığı buraya
if not data:
raise ValueError("Veri yükü boş")Asenkron Bağlamlarda Stratejik Hata Yönetimi
Asenkron Python kodu, hata yönetimi için benzersiz zorluklar getirir. Standart birtry-except bloğu tek iş parçacıklı yürütme için çalışır, ancak asyncio veya FastAPI gibi çerçevelerde, istisnalar olay döngüsü veya ara katman tarafından yakalanmazsa sessizce yayılabilir. Ayrıca, yüksek geçişli sistemlerde, hataları günlüklerken olay döngüsünü bloklamak, gecikme zincirine neden olabilir.
Üretim seviyesinde hata yönetimi iki farklı strateji gerektirir: **kategorizasyon** ve **bağlam yayılımı**.
1. Hata Kategorizasyonu
Gerekli olmadıkça genelException yakalamayın. Bunun yerine, belirli HTTP durum kodlarına veya yeniden deneme stratejilerine haritalanmış özel istisna hiyerarşileri oluşturun. Bu, ara katmanınızın iş mantığı hatalarını altyapı arızalarından farklı şekilde ele almasını sağlar.
class AppException(Exception):
def __init__(self, message, status_code, error_code=None):
super().__init__(message)
self.status_code = status_code
self.error_code = error_code
self.details = {}
class ValidationError(AppException):
def __init__(self, message):
super().__init__(message, status_code=400, error_code="VALIDATION_ERROR")
class RateLimitExceeded(AppException):
def __init__(self, limit):
super().__init__(f"{limit} oran sınırı aşıldı.", status_code=429, error_code="RATE_LIMIT")2. Bağlamsal Hata Yayılımı
Bir hata oluştuğunda, günlük kaydı, istisna yığın izi ve istek yaşam döngüsü boyunca toplanan yapılandırılmış veriler dahil olmak üzere istekle ilgili tam bağlamı içermelidir. Asenkron bir çerçevede, bu bağlamı günlükçüye enjekte eden küresel istisna işleyicilerinden yararlanmalısınız.import traceback
async def global_exception_handler(request, exc):
# Bu hatayı tetikleyen kullanıcıyı veya hizmeti belirleyin
user_id = getattr(request.state, 'user_id', 'unknown')
request_id = request.headers.get('x-request-id', 'no-id')
logger.error(
"İstekte yakalanamayan istisna",
extra={
"request_id": request_id,
"user_id": user_id,
"exception_type": type(exc).__name__,
"exception_message": str(exc),
"traceback": traceback.format_exc(),
"stack_level": "error"
}
)
# Standartlaştırılmış bir hata yanıtı döndürün
return {"error": "İç Sunucu Hatası", "trace_id": request_id}Performans İçin Optimize Etme
Yüksek eşzamanlılık uygulamalarında günlük kaydı, optimize edilmezse kendisi bir darboğaz haline gelebilir. Günlük çağrılarınızın içinde büyük dize nesneleri oluşturmaktan veya pahalı işlemler yapmaktan kaçının. Her zaman geç değerlendirmeyi kullanarak günlük yönteme argümanlar gönderin; bu, günlük seviyesi aktif olduğunda mesajın hesaplanmasını sağlar. Ayrıca, ağır XML ayrıştırıcılar yerineloguru kullanan veya optimize edilmiş JSON biçimlendiriciler kullanan yapılandırılmış günlük kütüphanelerini tercih edin.
Günlük çıktınızın asenkron olduğundan emin olun. Yüksek trafik patlamaları sırasında diske senkron olarak yazmak, çalışanlarınızı bloke edebilir. Günlük oluşturma ile günlük kalıcılığını ayırmak için yazmaları arabelleğe alan veya günlükleri arka plan bir iş parçacığına gönderen (örneğin QueueHandler) günlük işleyicilerini kullanın.