Locust ile Python Tabanlı Performans Testi

Yazar:komut.io ekibi
0 Yorum

Performans testine başlamak için JMeter’ı açıp karmaşık XML dosyalarıyla boğuşmak zorunda olmadığınızı fark ettiğinizde, Locust ile tanışma zamanınız gelmiş demektir. Yük testi dünyasına Python ile giriş yapmak, hem geliştirme ekiplerine yakın çalışan sysadmin’ler hem de CI/CD pipeline’larına test entegre etmek isteyen DevOps mühendisleri için gerçek bir kolaylık sağlıyor. Bu yazıda Locust’u sıfırdan kurarak, gerçek dünya senaryolarına uygun testler yazacağız ve distributed test koşumuna kadar götüreceğiz.

Locust Nedir ve Neden Tercih Edilmeli

Locust, Python ile yazılmış açık kaynaklı bir yük testi aracıdır. Kullanıcı davranışlarını saf Python koduyla tanımlarsınız; döngüler, koşullar, kütüphaneler, hepsi kullanılabilir. JMeter’ın GUI bağımlılığı veya k6’nın JavaScript zorunluluğu yoktur. Testlerinizi versiyon kontrol sistemine atmak, review’dan geçirmek ve pipeline’a entegre etmek çok daha doğal hale gelir.

Özellikle şu durumlarda Locust ön plana çıkıyor:

  • Python ekosistemiyle zaten çalışan takımlarda kod paylaşımı ve ortak utility kullanımı mümkün oluyor
  • Web UI üzerinden gerçek zamanlı metrik takibi yapılabiliyor
  • Distributed mod ile tek bir koordinatör node üzerinden onlarca worker’ı yönetebiliyorsunuz
  • Headless modda CI/CD sistemlerine kolayca entegre ediliyor

Kurulum ve İlk Adımlar

Locust’u kurmak gerçekten bu kadar basit:

pip install locust

# Versiyon kontrolü
locust --version

# Sanal ortam kullanıyorsanız (tavsiye edilir)
python -m venv locust-env
source locust-env/bin/activate
pip install locust
pip install locust[mqtt]  # MQTT protokolü için ek bağımlılıklar

Projenizi yapılandırırken bir requirements.txt tutmanızı öneririm. Özellikle CI ortamlarında versiyon uyumsuzlukları can sıkabilir:

# requirements.txt
locust==2.24.0
faker==24.0.0
requests==2.31.0

İlk Locustfile: Temel Yapı

Her şey locustfile.py dosyasıyla başlar. Locust çalıştırıldığında bu dosyayı otomatik olarak arar:

from locust import HttpUser, task, between
import json
import random

class WebsiteUser(HttpUser):
    # Her kullanıcı iki görev arasında 1-5 saniye bekler
    wait_time = between(1, 5)
    
    def on_start(self):
        """Kullanıcı simülasyonu başladığında çalışır - login gibi işlemler burada"""
        response = self.client.post("/api/auth/login", json={
            "username": "[email protected]",
            "password": "test_password_123"
        })
        
        if response.status_code == 200:
            token = response.json().get("access_token")
            # Sonraki isteklerde header olarak kullan
            self.client.headers.update({"Authorization": f"Bearer {token}"})
        else:
            # Login başarısız olursa kullanıcıyı durdur
            self.environment.runner.quit()
    
    @task(3)
    def view_products(self):
        """Ağırlıklı görev: ürün listesi görüntüleme"""
        self.client.get("/api/products?page=1&limit=20")
    
    @task(1)
    def view_single_product(self):
        """Daha az sıklıkta: tekil ürün sayfası"""
        product_id = random.randint(1, 500)
        self.client.get(f"/api/products/{product_id}")
    
    @task(1)
    def search_products(self):
        """Arama işlemi"""
        keywords = ["laptop", "telefon", "tablet", "kulaklık", "mouse"]
        keyword = random.choice(keywords)
        self.client.get(f"/api/search?q={keyword}")
    
    def on_stop(self):
        """Kullanıcı simülasyonu bittiğinde - logout"""
        self.client.post("/api/auth/logout")

@task dekoratöründeki sayı göreceli ağırlığı ifade eder. Yukarıdaki örnekte ürün listesi görüntüleme, diğer iki işlemden 3 kat daha sık gerçekleşir. Bu, gerçek kullanıcı davranışını modellemek için kritik bir detay.

E-Ticaret Senaryosu: Gerçek Dünya Örneği

Teorik örneklerden çıkıp gerçekçi bir senaryo kuralım. Bir e-ticaret platformunda “Black Friday” öncesi yük testi yapmamız gerekiyor. Kullanıcıların büyük çoğunluğu sadece geziniyor, küçük bir kısmı sepete ekliyor, daha da azı satın alıyor:

from locust import HttpUser, task, between, SequentialTaskSet
from faker import Faker
import json
import random

fake = Faker('tr_TR')

class ShoppingFlow(SequentialTaskSet):
    """
    SequentialTaskSet ile adımları sırasıyla çalıştırıyoruz.
    Gerçek bir alışveriş akışını simüle eder.
    """
    
    def on_start(self):
        self.cart_id = None
        self.selected_products = []
    
    @task
    def browse_homepage(self):
        with self.client.get("/", catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"Ana sayfa yüklenemedi: {response.status_code}")
    
    @task
    def browse_category(self):
        categories = ["elektronik", "giyim", "ev-yasam", "spor", "kitap"]
        category = random.choice(categories)
        
        with self.client.get(
            f"/kategori/{category}",
            name="/kategori/[name]",  # Locust UI'da gruplamak için
            catch_response=True
        ) as response:
            if response.status_code == 200:
                data = response.json()
                # Sonraki adımda kullanmak üzere ürün ID'lerini saklıyoruz
                if data.get("products"):
                    self.selected_products = [p["id"] for p in data["products"][:5]]
            else:
                response.failure("Kategori sayfası başarısız")
    
    @task
    def view_product_detail(self):
        if not self.selected_products:
            return
            
        product_id = random.choice(self.selected_products)
        
        with self.client.get(
            f"/urun/{product_id}",
            name="/urun/[id]",
            catch_response=True
        ) as response:
            if response.elapsed.total_seconds() > 2.0:
                response.failure(f"Ürün sayfası çok yavaş: {response.elapsed.total_seconds():.2f}s")
    
    @task
    def add_to_cart(self):
        if not self.selected_products:
            self.interrupt()
            return
            
        product_id = random.choice(self.selected_products)
        
        response = self.client.post("/sepet/ekle", json={
            "product_id": product_id,
            "quantity": random.randint(1, 3),
            "variant_id": None
        })
        
        if response.status_code == 200:
            self.cart_id = response.json().get("cart_id")
        
        # Sepete ekleme işleminden sonra akışı bitir
        self.interrupt()

class BrowsingUser(HttpUser):
    """Sadece gezen, satın almayan kullanıcılar - trafiğin %70'i"""
    tasks = [ShoppingFlow]
    wait_time = between(2, 6)
    weight = 7
    
    def on_start(self):
        # Anonim kullanıcı - login gerektirmez
        pass

class RegisteredUser(HttpUser):
    """Kayıtlı, alışveriş yapan kullanıcılar - trafiğin %30'u"""
    tasks = [ShoppingFlow]
    wait_time = between(1, 3)
    weight = 3
    
    def on_start(self):
        self.client.post("/giris", json={
            "email": fake.email(),
            "password": "Demo1234!"
        })

Custom Metrik Toplama

Locust’un varsayılan metrikleri çoğu zaman yetmez. Özellikle iş süreçlerine özel metrikleri takip etmek istiyorsunuz. Örneğin “ödeme sayfası yükleme süresi” ayrı bir metrik olarak raporlanabilir:

from locust import HttpUser, task, between, events
from locust.runners import MasterRunner
import time

# Custom event tanımlama
checkout_time_histogram = []

@events.request.add_listener
def on_request(request_type, name, response_time, response_length, 
               response, context, exception, **kwargs):
    """Her istek sonrası tetiklenir"""
    if name == "/odeme/tamamla" and exception is None:
        checkout_time_histogram.append(response_time)

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """Test bitişinde özet yazdır"""
    if checkout_time_histogram:
        avg = sum(checkout_time_histogram) / len(checkout_time_histogram)
        max_time = max(checkout_time_histogram)
        p95 = sorted(checkout_time_histogram)[int(len(checkout_time_histogram) * 0.95)]
        
        print(f"n=== Ödeme Akışı Özeti ===")
        print(f"Toplam işlem: {len(checkout_time_histogram)}")
        print(f"Ortalama süre: {avg:.0f}ms")
        print(f"P95 süre: {p95:.0f}ms")
        print(f"Maksimum süre: {max_time:.0f}ms")

class CheckoutUser(HttpUser):
    wait_time = between(2, 5)
    
    @task
    def complete_purchase(self):
        # Sepet oluştur
        self.client.post("/sepet/ekle", json={"product_id": 42, "quantity": 1})
        
        # Ödeme sayfası
        with self.client.get("/odeme", catch_response=True, name="/odeme/sayfa") as r:
            if r.status_code != 200:
                r.failure("Ödeme sayfası açılmadı")
                return
        
        # Ödemeyi tamamla - bu isteği özel olarak izliyoruz
        payment_data = {
            "card_number": "4242424242424242",
            "expiry": "12/25",
            "cvv": "123",
            "address_id": 1
        }
        
        start = time.time()
        response = self.client.post("/odeme/tamamla", json=payment_data)
        elapsed = (time.time() - start) * 1000
        
        if response.status_code == 200:
            print(f"Ödeme tamamlandı: {elapsed:.0f}ms")
        else:
            print(f"Ödeme başarısız: {response.status_code}")

Locust Komut Satırı Kullanımı

Web UI güzel ama production testlerinde ve CI/CD’de headless mod şart:

# Temel headless koşum: 100 kullanıcı, saniyede 10 kullanıcı spawn
locust -f locustfile.py 
  --headless 
  --users 100 
  --spawn-rate 10 
  --run-time 5m 
  --host https://staging.example.com 
  --csv=results/load_test_$(date +%Y%m%d_%H%M%S)

# Farklı bir locustfile kullanmak için
locust -f tests/performance/checkout_test.py 
  --headless 
  --users 500 
  --spawn-rate 50 
  --run-time 10m 
  --host https://api.example.com 
  --html=reports/report.html 
  --loglevel INFO

# Belirli kullanıcı sınıflarını çalıştırmak (birden fazla varsa)
locust -f locustfile.py 
  --headless 
  --users 200 
  --spawn-rate 20 
  --run-time 3m 
  --host https://staging.example.com 
  BrowsingUser RegisteredUser

# Başarısızlık eşiği belirlemek (CI için kritik)
locust -f locustfile.py 
  --headless 
  --users 100 
  --spawn-rate 10 
  --run-time 2m 
  --host https://staging.example.com 
  --exit-code-on-error 1 
  --stop-timeout 30

Distributed Mod: Master-Worker Mimarisi

Tek makine yetmediğinde distributed moda geçiyorsunuz. Gerçek bir yük testinde bunu mutlaka kullanmanız gerekiyor çünkü tek bir Locust instance’ı Python’ın GIL kısıtlaması nedeniyle belirli bir noktada darboğaza giriyor:

# Master node'u başlat
locust -f locustfile.py 
  --master 
  --master-bind-host=0.0.0.0 
  --master-bind-port=5557 
  --host https://production-staging.example.com 
  --users 2000 
  --spawn-rate 100

# Her worker node'da (farklı sunucular)
locust -f locustfile.py 
  --worker 
  --master-host=192.168.1.100 
  --master-port=5557

# Docker Compose ile basit distributed setup
# docker-compose.yml içeriği aşağıda
# docker-compose.yml
version: '3.8'

services:
  locust-master:
    image: locustio/locust:2.24.0
    ports:
      - "8089:8089"
      - "5557:5557"
    volumes:
      - ./tests:/mnt/locust
    command: >
      -f /mnt/locust/locustfile.py
      --master
      --host https://staging.example.com
    environment:
      - LOCUST_USERS=1000
      - LOCUST_SPAWN_RATE=50

  locust-worker:
    image: locustio/locust:2.24.0
    volumes:
      - ./tests:/mnt/locust
    command: >
      -f /mnt/locust/locustfile.py
      --worker
      --master-host=locust-master
    depends_on:
      - locust-master
    deploy:
      replicas: 4  # 4 worker container

CI/CD Entegrasyonu: GitLab CI Örneği

Staging ortamına her deployment sonrası otomatik performans testi çalıştırmak gerçekten hayat kurtarıcı. Biz bir projede bu yaklaşımla production’a geçmeden önce bir N+1 sorgusu kaynaklı yavaşlamayı yakalamıştık:

# .gitlab-ci.yml
stages:
  - test
  - performance

performance-test:
  stage: performance
  image: python:3.11-slim
  only:
    - staging
  before_script:
    - pip install locust faker --quiet
  script:
    - |
      locust -f tests/performance/locustfile.py 
        --headless 
        --users 50 
        --spawn-rate 5 
        --run-time 3m 
        --host $STAGING_URL 
        --csv=perf-results/test 
        --html=perf-results/report.html 
        --exit-code-on-error 1
    
    # P95 yanıt süresini kontrol et
    - python tests/performance/check_results.py perf-results/test_stats.csv
  artifacts:
    when: always
    paths:
      - perf-results/
    reports:
      junit: perf-results/junit.xml
  allow_failure: false
# check_results.py - Sonuçları değerlendirme scripti
import csv
import sys

def check_performance_results(csv_file):
    thresholds = {
        "p95_response_time_ms": 2000,  # P95 2 saniyeden az olmalı
        "failure_rate_percent": 1.0,    # Hata oranı %1'den az olmalı
        "min_rps": 50                   # Minimum 50 istek/saniye olmalı
    }
    
    failures = []
    
    with open(csv_file, newline='') as f:
        reader = csv.DictReader(f)
        for row in reader:
            if row['Name'] == 'Aggregated':
                p95 = float(row.get('95%', 0))
                failure_count = int(row.get('Failure Count', 0))
                request_count = int(row.get('Request Count', 1))
                rps = float(row.get('Requests/s', 0))
                
                failure_rate = (failure_count / request_count) * 100
                
                if p95 > thresholds["p95_response_time_ms"]:
                    failures.append(f"P95 yanıt süresi çok yüksek: {p95:.0f}ms (eşik: {thresholds['p95_response_time_ms']}ms)")
                
                if failure_rate > thresholds["failure_rate_percent"]:
                    failures.append(f"Hata oranı yüksek: {failure_rate:.2f}% (eşik: {thresholds['failure_rate_percent']}%)")
                
                if rps < thresholds["min_rps"]:
                    failures.append(f"Throughput düşük: {rps:.1f} RPS (minimum: {thresholds['min_rps']})")
    
    if failures:
        print("PERFORMANS TESTİ BAŞARISIZ:")
        for f in failures:
            print(f"  - {f}")
        sys.exit(1)
    else:
        print("Performans testi başarıyla geçti.")
        sys.exit(0)

if __name__ == "__main__":
    check_performance_results(sys.argv[1])

Sonuç Yorumlama ve Dikkat Edilmesi Gerekenler

Locust sonuçlarını okurken yanılabileceğiniz birkaç nokta var:

Ortalama yanıt süresi aldatıcı olabilir. P50, P95 ve P99 değerlerine bakın. Ortalama 200ms olan bir endpoint’in P99’u 8 saniye olabilir ve bu kullanıcıların %1’inin berbat bir deneyim yaşadığı anlamına gelir.

Spawn rate kritik bir parametredir. Çok hızlı kullanıcı eklemek, sunucuya anlık bir spike yaşatır ve gerçek yük profilini bozar. E-ticaret siteleri için genellikle kullanıcı sayısının %5-10’u kadar bir spawn rate mantıklıdır.

Test ortamı production’ı yansıtmalı. Staging’de 4 CPU’luk sunucu var ama production 16 CPU ise, test sonuçları yanıltıcı olur. Mümkünse production-like bir ortamda test yapın.

Locust’un kendisi darboğaz olabilir. 10.000 kullanıcı üzerinde single instance çalıştırırsanız, Locust’un kaynak tüketimi ölçümlerinizi kirletir. Distributed mod bu yüzden sadece büyük testler için değil, sonuçların güvenilirliği için de önemlidir.

Bağımlı servisleri unutmayın. Uygulamanız veritabanı, Redis, dış API çağırdığında bunlar da teste dahil olur. Yavaş bir test sonucu her zaman web katmanı kaynaklı değildir. Locust testi sırasında veritabanı slow query log’larını da izleyin.

Locust’u bir kez doğru kurguladığınızda, ekibin genel sağlık kontrolü haline gelir. Her sprint sonunda otomatik koşan 5 dakikalık bir yük testi bile, performance regression’ları erkenden yakalamanızı sağlar. Başlamak için mükemmel bir senaryo beklemeyin; basit bir endpoint testi bile sizi doğru yöne götürür.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir