API Bağlantı Havuzu: Connection Pooling ile Performans Optimizasyonu

Tarafındankomut
0 Comments

Üretim ortamında her saniye binlerce API isteği işleyen bir sistem yönetiyorsunuz ve aniden response time’larınız tavan yapıyor. Logları inceliyorsunuz, CPU ve RAM normal, network de sorun değil. Derinden bakınca sorunun her istek için yeni bir TCP bağlantısı açılıp kapatılmasından kaynaklandığını görüyorsunuz. İşte bu klasik senaryo, connection pooling’in neden bu kadar kritik olduğunu anlatıyor.

Connection Pooling Nedir ve Neden Önemlidir

API geliştirme dünyasında en sık yapılan hatalardan biri, her HTTP isteği için yeni bir bağlantı kurulması. TCP handshake, TLS negotiation, DNS çözümleme… Bunların her biri ciddi gecikme demek. Düşük trafikte fark edilmez ama yük arttığında sistem çöküşe geçer.

Connection pooling, önceden belirli sayıda bağlantı açarak bunları bir havuzda tutma ve gelen isteklere bu hazır bağlantıları tahsis etme tekniğidir. İstek tamamlandığında bağlantı kapatılmaz, havuza geri döner ve bir sonraki istek tarafından kullanılır.

Bir bağlantı kurmanın maliyetine bakalım:

  • DNS çözümleme: 20-120ms
  • TCP 3-way handshake: 1 RTT (round trip time)
  • TLS 1.2 handshake: 2 RTT
  • TLS 1.3 handshake: 1 RTT (0-RTT ile daha da az)
  • HTTP request/response: asıl iş bu

Saniyede 1000 istek gelen bir sistemde her seferinde bu maliyeti ödemek, gereksiz yük oluşturur ve latency’yi ciddi artırır.

Temel Kavramlar

Pool Size ve Workers İlişkisi

Connection pool boyutunu belirlerken uygulamanızın concurrency modelini anlamanız gerekir. Yanlış yapılandırılmış bir pool, yokluğundan daha zararlı olabilir.

  • min_connections: Havuzda her zaman hazır bekleyen minimum bağlantı sayısı
  • max_connections: Havuzun açabileceği maksimum bağlantı sayısı
  • idle_timeout: Boşta kalan bağlantının kapatılma süresi
  • connection_timeout: Yeni bağlantı açarken beklenecek maksimum süre
  • max_lifetime: Bir bağlantının havuzda kalabileceği maksimum ömür

Genel kural olarak: pool_size = (core_count * 2) + effective_spindle_count formülü iyi bir başlangıç noktasıdır. Ama bu kesin değil, yük testleriyle fine-tune edilmesi gerekir.

Keep-Alive vs Connection Pooling Farkı

HTTP Keep-Alive, aynı bağlantı üzerinden birden fazla istek göndermeyi sağlar ama uygulama katmanında bağlantı yönetimi yoktur. Connection pooling ise uygulama seviyesinde aktif bağlantı yönetimi demektir: bağlantı durumları takip edilir, sağlık kontrolleri yapılır, maksimum kullanım limitleri uygulanır.

Python ile Connection Pooling

Python ekosisteminde requests kütüphanesi çok popüler ama her istek için session kullanmadan çağrı yapmak büyük hata. httpx ve aiohttp ise async ortamlar için güçlü alternatifler.

requests.Session ile Temel Pooling

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session(
    pool_connections=10,
    pool_maxsize=20,
    max_retries=3,
    backoff_factor=0.5
):
    session = requests.Session()
    
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "POST", "PUT", "DELETE"]
    )
    
    adapter = HTTPAdapter(
        pool_connections=pool_connections,  # farklı host'lar için pool sayisi
        pool_maxsize=pool_maxsize,           # her pool'daki max baglanti
        max_retries=retry_strategy,
        pool_block=False  # pool dolu olunca exception firlatir, beklemez
    )
    
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    # varsayilan timeout ayarla
    session.request = lambda method, url, **kwargs: 
        requests.Session.request(session, method, url, 
                                  timeout=kwargs.pop('timeout', (5, 30)), 
                                  **kwargs)
    return session

# kullanim
api_session = create_session(pool_connections=5, pool_maxsize=15)
response = api_session.get("https://api.ornekservis.com/v1/users")

httpx ile Async Connection Pool

import asyncio
import httpx
from contextlib import asynccontextmanager

# uygulama omru boyunca yasayan tek bir client instance
_client = None

@asynccontextmanager
async def get_http_client():
    global _client
    if _client is None:
        limits = httpx.Limits(
            max_keepalive_connections=20,
            max_connections=50,
            keepalive_expiry=30.0  # saniye
        )
        timeout = httpx.Timeout(
            connect=5.0,
            read=30.0,
            write=10.0,
            pool=5.0  # pool'dan baglanti almak icin bekleme suresi
        )
        _client = httpx.AsyncClient(
            limits=limits,
            timeout=timeout,
            http2=True  # HTTP/2 aktif, daha verimli multiplexing
        )
    try:
        yield _client
    except Exception as e:
        raise e

async def fetch_user_data(user_id: int):
    async with get_http_client() as client:
        response = await client.get(
            f"https://api.ornekservis.com/v1/users/{user_id}"
        )
        response.raise_for_status()
        return response.json()

async def bulk_fetch(user_ids: list):
    async with get_http_client() as client:
        tasks = [
            client.get(f"https://api.ornekservis.com/v1/users/{uid}")
            for uid in user_ids
        ]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        results = []
        for resp in responses:
            if isinstance(resp, Exception):
                results.append({"error": str(resp)})
            else:
                results.append(resp.json())
        return results

Node.js ile Connection Pooling

Node.js’in event loop mimarisi connection pooling için ideal bir zemin sunar. axios ve node-fetch kullanırken dikkat edilmesi gereken bazı noktalar var.

axios ile Global Agent Konfigürasyonu

const axios = require('axios');
const http = require('http');
const https = require('https');

// varsayilan agent'lari override et
const httpAgent = new http.Agent({
    keepAlive: true,
    keepAliveMsecs: 30000,   // 30 saniye keep-alive
    maxSockets: 50,           // host basina max socket
    maxFreeSockets: 20,       // bosta bekleyen max socket
    timeout: 60000            // socket timeout
});

const httpsAgent = new https.Agent({
    keepAlive: true,
    keepAliveMsecs: 30000,
    maxSockets: 50,
    maxFreeSockets: 20,
    timeout: 60000,
    rejectUnauthorized: true  // SSL dogrulama acik kalsin
});

const apiClient = axios.create({
    baseURL: 'https://api.ornekservis.com/v1',
    httpAgent,
    httpsAgent,
    timeout: 30000,
    headers: {
        'Content-Type': 'application/json',
        'Accept-Encoding': 'gzip, deflate, br'
    }
});

// interceptor ile merkezi hata yonetimi
apiClient.interceptors.response.use(
    response => response,
    async error => {
        const { config, response } = error;
        
        if (!config || !config.retryCount) {
            config.retryCount = 0;
        }
        
        const retryableStatuses = [429, 500, 502, 503, 504];
        if (
            config.retryCount < 3 &&
            response &&
            retryableStatuses.includes(response.status)
        ) {
            config.retryCount++;
            const delay = Math.pow(2, config.retryCount) * 1000;
            await new Promise(resolve => setTimeout(resolve, delay));
            return apiClient(config);
        }
        
        return Promise.reject(error);
    }
);

module.exports = apiClient;

Go ile Connection Pooling

Go’nun standart kütüphanesi connection pooling için oldukça güçlü araçlar sunar. http.Transport yapısı doğrudan havuz konfigürasyonuna izin verir.

package main

import (
    "context"
    "fmt"
    "net"
    "net/http"
    "time"
)

func createOptimizedClient() *http.Client {
    dialer := &net.Dialer{
        Timeout:   5 * time.Second,   // baglanti kurma timeout
        KeepAlive: 30 * time.Second,  // TCP keep-alive interval
    }

    transport := &http.Transport{
        DialContext:           dialer.DialContext,
        MaxIdleConns:          100,             // toplam max idle baglanti
        MaxIdleConnsPerHost:   20,              // host basina max idle
        MaxConnsPerHost:       50,              // host basina max toplam
        IdleConnTimeout:       90 * time.Second,
        TLSHandshakeTimeout:   10 * time.Second,
        ExpectContinueTimeout: 1 * time.Second,
        DisableCompression:    false,
        ForceAttemptHTTP2:     true,
    }

    return &http.Client{
        Transport: transport,
        Timeout:   30 * time.Second,
    }
}

// singleton pattern ile global client
var globalClient = createOptimizedClient()

func fetchWithPool(ctx context.Context, url string) ([]byte, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return nil, fmt.Errorf("request olusturulamadi: %w", err)
    }

    req.Header.Set("Accept-Encoding", "gzip")
    
    resp, err := globalClient.Do(req)
    if err != nil {
        return nil, fmt.Errorf("istek basarisiz: %w", err)
    }
    defer resp.Body.Close()

    // response body okuma...
    return nil, nil
}

Gerçek Dünya Senaryosu: Mikro Servis Mimarisinde Pool Yönetimi

Bir e-ticaret platformunda çalıştığınızı düşünün. Order servisi, payment servisi ve inventory servisiyle konuşuyor. Her servise ayrı pool konfigürasyonu gerekiyor çünkü kritiklik seviyeleri ve response time beklentileri farklı.

import httpx
from dataclasses import dataclass
from typing import Dict

@dataclass
class ServiceConfig:
    base_url: str
    max_connections: int
    max_keepalive: int
    timeout_connect: float
    timeout_read: float

# servis bazinda farkli pool konfigurasyonlari
SERVICE_CONFIGS: Dict[str, ServiceConfig] = {
    "payment": ServiceConfig(
        base_url="https://payment.internal:8443",
        max_connections=10,      # kritik servis, kontrollü erisim
        max_keepalive=5,
        timeout_connect=2.0,
        timeout_read=15.0        # odeme islemi uzun surebilir
    ),
    "inventory": ServiceConfig(
        base_url="https://inventory.internal:8080",
        max_connections=30,      # yuksek frekansta sorgulanir
        max_keepalive=15,
        timeout_connect=1.0,
        timeout_read=5.0
    ),
    "notification": ServiceConfig(
        base_url="https://notification.internal:8080",
        max_connections=50,      # fire-and-forget, yuksek hacim
        max_keepalive=25,
        timeout_connect=1.0,
        timeout_read=3.0
    )
}

class ServiceClientPool:
    def __init__(self):
        self._clients: Dict[str, httpx.AsyncClient] = {}
    
    async def get_client(self, service_name: str) -> httpx.AsyncClient:
        if service_name not in self._clients:
            config = SERVICE_CONFIGS[service_name]
            limits = httpx.Limits(
                max_connections=config.max_connections,
                max_keepalive_connections=config.max_keepalive
            )
            timeout = httpx.Timeout(
                connect=config.timeout_connect,
                read=config.timeout_read
            )
            self._clients[service_name] = httpx.AsyncClient(
                base_url=config.base_url,
                limits=limits,
                timeout=timeout,
                verify="/etc/ssl/certs/internal-ca.pem"
            )
        return self._clients[service_name]
    
    async def close_all(self):
        for client in self._clients.values():
            await client.aclose()
        self._clients.clear()

# uygulama startup/shutdown ile entegrasyon (FastAPI ornegi)
from fastapi import FastAPI

app = FastAPI()
client_pool = ServiceClientPool()

@app.on_event("startup")
async def startup():
    # client'lari onceden olustur, ilk istek gecikmesi olmasin
    for service in SERVICE_CONFIGS:
        await client_pool.get_client(service)

@app.on_event("shutdown")
async def shutdown():
    await client_pool.close_all()

Pool Sağlığını İzleme ve Metrikler

Connection pool’u yapılandırmak yetmez, izlemeniz de gerekir. Prometheus ile entegrasyon şart.

from prometheus_client import Gauge, Counter, Histogram
import threading

# prometheus metrikleri tanimla
pool_active_connections = Gauge(
    'http_pool_active_connections',
    'Havuzda aktif olarak kullanimda olan baglanti sayisi',
    ['service', 'host']
)

pool_idle_connections = Gauge(
    'http_pool_idle_connections', 
    'Havuzda bosta bekleyen baglanti sayisi',
    ['service', 'host']
)

pool_wait_duration = Histogram(
    'http_pool_wait_seconds',
    'Pool'dan baglanti almak icin beklenen sure',
    ['service'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0]
)

connection_errors_total = Counter(
    'http_connection_errors_total',
    'Baglanti hatalarinin toplam sayisi',
    ['service', 'error_type']
)

class MonitoredHTTPAdapter(HTTPAdapter):
    """requests adapter'ini izleme ozellikleriyle genislet"""
    
    def __init__(self, service_name: str, *args, **kwargs):
        self.service_name = service_name
        super().__init__(*args, **kwargs)
    
    def send(self, request, *args, **kwargs):
        import time
        start = time.time()
        
        try:
            response = super().send(request, *args, **kwargs)
            return response
        except Exception as e:
            error_type = type(e).__name__
            connection_errors_total.labels(
                service=self.service_name,
                error_type=error_type
            ).inc()
            raise
        finally:
            # pool istatistiklerini guncelle
            self._update_pool_metrics()
    
    def _update_pool_metrics(self):
        """urllib3 pool manager'dan istatistikleri cek"""
        try:
            for host, pool in self.poolmanager.connection_pool_kw.items():
                pass  # gercek implementasyonda pool.pool.qsize() kullanilir
        except Exception:
            pass

Yaygın Hatalar ve Çözümleri

Connection Leak Problemi

En tehlikeli sorunlardan biri. Response body okunmadan bağlantı havuza dönmez.

import requests

# YANLIS - response body okunmuyor, baglanti leak oluyor
def hatali_kullanim(session):
    response = session.get("https://api.example.com/data")
    if response.status_code == 200:
        return True  # body okunmadi!

# DOGRU - context manager kullan
def dogru_kullanim(session):
    with session.get("https://api.example.com/data", stream=True) as response:
        response.raise_for_status()
        return response.json()

# DOGRU - streaming icin iter_content kullan
def streaming_kullanim(session):
    with session.get("https://api.example.com/largefile", stream=True) as response:
        response.raise_for_status()
        chunks = []
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                chunks.append(chunk)
        return b"".join(chunks)

Pool Tükenmesi ve Circuit Breaker

Pool tamamen dolduğunda ne olacak? Circuit breaker pattern bu durumu yönetir.

import time
from enum import Enum
from threading import Lock

class CircuitState(Enum):
    CLOSED = "closed"       # normal calisma
    OPEN = "open"           # servis kapali, istek gecme
    HALF_OPEN = "half_open" # test asamasi

class CircuitBreaker:
    def __init__(
        self, 
        failure_threshold=5,
        recovery_timeout=60,
        half_open_max_calls=3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0
        self._lock = Lock()
    
    def call(self, func, *args, **kwargs):
        with self._lock:
            if self.state == CircuitState.OPEN:
                elapsed = time.time() - self.last_failure_time
                if elapsed > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                else:
                    raise Exception(
                        f"Circuit breaker OPEN durumda. "
                        f"{self.recovery_timeout - elapsed:.1f}s sonra tekrar dene."
                    )
            
            if self.state == CircuitState.HALF_OPEN:
                if self.half_open_calls >= self.half_open_max_calls:
                    raise Exception("Half-open limit asildi")
                self.half_open_calls += 1
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        with self._lock:
            self.failure_count = 0
            self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

Nginx ve HAProxy ile Upstream Pool Yönetimi

Uygulama katmanı dışında, reverse proxy seviyesinde de pool yönetimi kritik.

# nginx upstream keepalive konfigurasyonu
upstream backend_api {
    server api1.internal:8080 weight=3 max_fails=3 fail_timeout=30s;
    server api2.internal:8080 weight=3 max_fails=3 fail_timeout=30s;
    server api3.internal:8080 weight=2 max_fails=3 fail_timeout=30s;
    
    # her upstream worker'i icin keepalive baglanti sayisi
    keepalive 32;
    
    # baglanti canliligi icin bekleme suresi
    keepalive_timeout 65s;
    
    # her baglanti uzerinden kac istek gecsin
    keepalive_requests 1000;
}

server {
    listen 443 ssl http2;
    server_name api.orneksirket.com;
    
    location /v1/ {
        proxy_pass http://backend_api;
        
        # HTTP/1.1 zorunlu, keepalive icin
        proxy_http_version 1.1;
        proxy_set_header Connection "";  # connection header'i temizle
        
        proxy_connect_timeout 5s;
        proxy_send_timeout 30s;
        proxy_read_timeout 30s;
        
        proxy_buffer_size 4k;
        proxy_buffers 8 4k;
    }
}

Performans Testleri ile Pool Boyutunu Belirlemek

Teorik hesaplamalar yetmez, gerçek yük testleri yapmanız şart.

#!/bin/bash
# wrk ile farkli pool boyutlarini test et

# parametreler
TARGET_URL="https://api.orneksirket.com/v1/health"
DURATION="30s"
THREADS=4

echo "=== Connection Pool Performans Testi ==="
echo "Tarih: $(date)"
echo "Hedef: $TARGET_URL"
echo ""

# farkli concurrent baglanti sayilariyla test
for CONNECTIONS in 10 25 50 100 200; do
    echo "--- Concurrent: $CONNECTIONS baglanti ---"
    
    # wrk ile test calistir
    wrk -t$THREADS -c$CONNECTIONS -d$DURATION 
        --latency 
        --timeout 10s 
        -H "Authorization: Bearer test-token" 
        $TARGET_URL
    
    echo ""
    sleep 5  # servisin toparlanmasi icin bekle
done

# sonuclari analiz et
echo "Optimum pool boyutunu belirlemek icin:"
echo "- En dusuk p99 latency degeri hangi connection sayisinda?"
echo "- Requests/sec hangi noktadan sonra plato yapıyor?"
echo "- Hata oranı hangi noktada yükselmeye başlıyor?"

Sonuç

Connection pooling, yüksek trafikli API sistemlerinde temel bir optimizasyon değil zorunluluktur. Doğru yapılandırılmış bir pool, latency’yi dramatik biçimde düşürür, sunucu kaynaklarını korur ve sistemin yük altında kararlı kalmasını sağlar.

Özetleyecek olursak:

  • Her zaman session veya client nesnelerini yeniden kullanın. Her istek için yeni nesne oluşturmak birincil hata kaynağıdır.
  • Pool boyutunu iş yüküne göre ölçün. Core count formülü başlangıç noktası, gerçek ölçüm hedef.
  • Timeout değerlerini servis SLA’larına göre ayarlayın. Ödeme servisi ile bildirim servisi aynı timeout’u hak etmiyor.
  • Circuit breaker ile pool tükenmesine karşı savunma katmanı ekleyin. Havuz dolduğunda sistem cascade failure’a gitmemeli.
  • Prometheus ile pool metriklerini izleyin. Aktif bağlantı sayısı, bekleme süresi ve hata oranı temel göstergeniz olsun.
  • Nginx gibi reverse proxy’lerde upstream keepalive’ı unutmayın. Uygulama katmanında ne kadar optimize ederseniz edin, proxy katmanı düzgün ayarlanmamışsa kazanç yarıda kalır.

Bu teknikleri uyguladıktan sonra yük testleri yapın, metrikleri karşılaştırın ve ortamınıza özel optimal değerleri bulun. Connection pooling tek seferlik yapılandırma değil, sistemin büyümesiyle birlikte sürekli gözden geçirilmesi gereken bir konudur.

Bir yanıt yazın

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