Testing

Fault Injection Patterns for Microservices: Latency, Errors, and Resource Exhaustion

HelpMeTest

HelpMeTest

— 6 min read
Fault Injection Patterns for Microservices: Latency, Errors, and Resource Exhaustion

Fault injection is the practice of deliberately introducing failures into a system to verify that it handles them correctly. Unlike chaos engineering's broad experiments, fault injection is surgical — you inject a specific failure into a specific component and observe what happens. This guide covers the core patterns for microservice architectures.

Why Fault Injection?

Microservices have unique failure modes:

  • A service calls 5 downstream services. Any one can fail.
  • Network calls add latency variability.
  • Dependencies share infrastructure (databases, queues).
  • Cascading failures: one slow service causes timeouts upstream.

You can't test all combinations manually. Fault injection lets you systematically verify that each failure mode is handled correctly.

The four questions fault injection answers:

  1. Does the system degrade gracefully or fail catastrophically?
  2. Do circuit breakers trip when they should?
  3. Do retry mechanisms cause thundering herds?
  4. Does the monitoring detect the failure correctly?

Latency Injection

The most common production failure is not errors — it's slowness. A service that normally responds in 50ms starts taking 2 seconds. Everything upstream accumulates timeouts.

HTTP Latency with Toxiproxy

Toxiproxy is a proxy that sits between services and introduces controlled network failures:

# Install
docker run -d -p 8474:8474 -p 8080:8080 shopify/toxiproxy

# Create a proxy
curl -X POST http://localhost:8474/proxies \
  -H "Content-Type: application/json" \
  -d '{
    "name": "payment-service",
    "listen": "0.0.0.0:8080",
    "upstream": "payment-service:3000",
    "enabled": true
  }'

# Add latency toxic (200ms latency, 50ms jitter)
curl -X POST http://localhost:8474/proxies/payment-service/toxics \
  -H "Content-Type: application/json" \
  -d '{
    "name": "slow_payment",
    "type": "latency",
    "stream": "downstream",
    "attributes": {
      "latency": 200,
      "jitter": 50
    }
  }'

Now traffic through port 8080 → payment-service gets 200±50ms of added latency.

Testing Timeout Behavior

# test_payment_timeout.py
import requests
import toxiproxy

client = toxiproxy.Toxiproxy()

def test_checkout_handles_payment_timeout():
    """
    Verify: when payment-service is slow (> 500ms), 
    checkout returns 503 with retry-after header within 1 second.
    """
    proxy = client.get_proxy('payment-service')
    
    with proxy.toxics.add('timeout_sim', 'latency', stream='downstream',
                           attributes={'latency': 600, 'jitter': 0}):
        response = requests.post('http://checkout:8000/api/checkout', 
                                  json={'cart_id': 'test-123'},
                                  timeout=2.0)
        
        # Should fail gracefully, not timeout the HTTP request
        assert response.status_code == 503
        assert 'Retry-After' in response.headers
        assert response.elapsed.total_seconds() < 1.0  # fails fast
        
        body = response.json()
        assert body['error'] == 'payment_service_unavailable'

Latency Patterns to Test

Pattern Config What it tests
Constant latency 500ms, 0 jitter Timeout thresholds
Variable latency 200ms, 100ms jitter P99 handling
Gradual increase Start 50ms, ramp to 2000ms Circuit breaker trip point
Periodic spikes 50ms normally, 1000ms 10% Retry with jitter

Error Injection

HTTP Error Rates

# 30% of requests return 503
curl -X POST http://localhost:8474/proxies/payment-service/toxics \
  -H "Content-Type: application/json" \
  -d '{
    "name": "error_rate",
    "type": "limit_data",
    "stream": "downstream",
    "toxicity": 0.30
  }'

Or using service mesh (Istio):

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: payment-service
spec:
  hosts:
    - payment-service
  http:
    - fault:
        abort:
          httpStatus: 503
          percentage:
            value: 30.0
      route:
        - destination:
            host: payment-service

Testing Circuit Breaker Behavior

A circuit breaker should open after N failures and stop sending requests:

def test_circuit_breaker_opens_on_errors():
    """
    Verify: after 5 consecutive 503s from payment-service,
    checkout opens circuit breaker and stops calling payment-service.
    """
    call_count_before = metrics.get('payment_service.calls.total')
    
    # Inject 100% errors
    with inject_errors(service='payment-service', error_rate=1.0, status=503):
        # First 5 requests should fail (circuit breaker threshold)
        for i in range(5):
            response = requests.post('http://checkout/api/checkout', 
                                     json={'cart_id': f'test-{i}'})
            assert response.status_code in [503, 200]  # may be fallback
        
        # After threshold, should fail fast (circuit open)
        response = requests.post('http://checkout/api/checkout',
                                  json={'cart_id': 'test-circuit-open'})
        
        call_count_after = metrics.get('payment_service.calls.total')
        new_calls = call_count_after - call_count_before
        
        # Circuit breaker should have stopped forwarding after 5 failures
        assert new_calls <= 5, f"Circuit breaker didn't open: {new_calls} calls made"
        
        # Response should indicate circuit is open, not generic 503
        body = response.json()
        assert body['error'] in ['circuit_open', 'payment_service_unavailable']

def test_circuit_breaker_recovers():
    """
    Verify: after circuit opens, it tries again after 30 seconds (half-open),
    and closes if the call succeeds.
    """
    with inject_errors(service='payment-service', error_rate=1.0, status=503):
        # Open the circuit
        trigger_circuit_open('payment-service')
    
    # Wait for half-open timeout
    time.sleep(35)  # circuit's recovery timeout is 30s
    
    # Next request should succeed (circuit half-open → closed)
    response = requests.post('http://checkout/api/checkout',
                              json={'cart_id': 'recovery-test'})
    assert response.status_code == 200
    
    # Circuit should be fully closed now
    assert circuit_state('payment-service') == 'closed'

Resource Exhaustion

Connection Pool Exhaustion

def test_database_connection_exhaustion():
    """
    Verify: when all DB connections are held, new requests get
    queued (not dropped) and complete within SLO after connections free.
    """
    # Hold all connections (pool size = 20)
    held_connections = []
    for _ in range(20):
        conn = db.get_connection()
        held_connections.append(conn)
    
    # New request should queue, not fail immediately
    start = time.time()
    response = requests.get('http://api/users/1', timeout=5.0)
    elapsed = time.time() - start
    
    # Release connections
    for conn in held_connections:
        db.release_connection(conn)
    
    # Should succeed eventually, not fail with 500
    assert response.status_code == 200
    # SLO: complete within 3 seconds even under pool exhaustion
    assert elapsed < 3.0

Memory Pressure

import subprocess

def test_service_handles_memory_pressure():
    """
    Verify: under memory pressure, service degrades gracefully
    and recovers when pressure is released.
    """
    service_pid = get_service_pid('api-gateway')
    
    # Stress memory (fill 80% of system memory)
    stress = subprocess.Popen(['stress-ng', '--vm', '1', '--vm-bytes', '80%'])
    
    try:
        # Service should still respond (possibly slower)
        response = requests.get('http://api-gateway/health', timeout=5.0)
        assert response.status_code == 200
        
        # Response time may be higher under memory pressure
        assert response.elapsed.total_seconds() < 2.0  # degraded SLO
        
    finally:
        stress.terminate()
    
    # After pressure released, should return to normal
    time.sleep(5)
    response = requests.get('http://api-gateway/health')
    assert response.status_code == 200
    assert response.elapsed.total_seconds() < 0.2  # normal SLO

Network Partition Simulation

Test split-brain scenarios:

# Block traffic between service A and service B
# (Linux iptables approach)
iptables -A INPUT -s service-b-ip -j DROP
iptables -A OUTPUT -d service-b-ip -j DROP

# Verify service A detects the partition
curl http://service-a/health

# Restore
iptables -D INPUT -s service-b-ip -j DROP
iptables -D OUTPUT -d service-b-ip -j DROP

Using Chaos Mesh for Kubernetes:

apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: partition-payment-from-inventory
spec:
  action: partition
  mode: all
  selector:
    namespaces:
      - production
    labelSelectors:
      app: payment-service
  direction: both
  target:
    mode: all
    selector:
      namespaces:
        - production
      labelSelectors:
        app: inventory-service
  duration: "60s"

What to verify during partition:

  • Does payment-service degrade gracefully without inventory?
  • Does it use cached data or fail open/closed?
  • Are the right alerts firing?
  • Does it recover automatically when partition heals?

Dependency Failure Injection

Third-Party API Failures

Test behavior when external APIs (Stripe, SendGrid, etc.) fail:

from unittest.mock import patch
import responses

@responses.activate
def test_order_completes_when_email_fails():
    """
    Verify: order creation succeeds even if confirmation email fails.
    Email is best-effort, not part of the critical path.
    """
    # Mock successful payment
    responses.add(
        responses.POST,
        'https://api.stripe.com/v1/charges',
        json={'id': 'ch_test123', 'status': 'succeeded'},
        status=200
    )
    
    # Mock email failure
    responses.add(
        responses.POST,
        'https://api.sendgrid.com/v3/mail/send',
        json={'errors': [{'message': 'Service unavailable'}]},
        status=503
    )
    
    # Order should complete despite email failure
    response = requests.post('http://api/orders', json={
        'cart_id': 'test-cart',
        'email': 'customer@example.com'
    })
    
    assert response.status_code == 201
    order = response.json()
    assert order['status'] == 'confirmed'
    
    # Email error should be logged but not bubble up
    logs = get_service_logs('order-service', last_seconds=5)
    assert any('sendgrid' in log.lower() and 'error' in log.lower() 
                for log in logs)

DNS Failure

def test_service_handles_dns_failure():
    """
    Verify: when DNS resolution fails for upstream service,
    the service uses cached DNS entries and degrades gracefully.
    """
    # Block DNS resolution (iptables block port 53)
    block_dns()
    
    try:
        # Should work using DNS cache for first N seconds
        response = requests.get('http://api/products')
        assert response.status_code == 200
    finally:
        unblock_dns()

Observability Verification

Fault injection is only half the value — the other half is verifying that your monitoring catches it:

def test_alert_fires_on_error_injection():
    """
    Verify: when error rate > 1%, the HighErrorRate alert fires
    within 5 minutes.
    """
    # Clear any existing alerts
    alert_manager.clear_alerts()
    
    # Inject 5% errors
    with inject_errors(service='api-gateway', error_rate=0.05):
        # Generate traffic to trigger the error rate
        for _ in range(200):
            requests.get('http://api-gateway/api/products')
        
        # Wait for alert to fire (polling)
        alert = wait_for_alert('HighErrorRate', timeout=300)
    
    assert alert is not None, "HighErrorRate alert did not fire"
    assert alert['labels']['severity'] == 'critical'
    assert alert['labels']['service'] == 'api-gateway'
    
    # Alert should resolve after injection stops
    alert_resolved = wait_for_alert_resolution('HighErrorRate', timeout=300)
    assert alert_resolved, "Alert did not resolve after injection stopped"

Combining Fault Injection into Test Suites

Structure fault injection tests as a dedicated test suite:

# tests/resilience/test_payment_resilience.py
import pytest

@pytest.mark.resilience
class TestPaymentServiceResilience:
    
    @pytest.fixture(autouse=True)
    def reset_circuit_breakers(self):
        """Reset all circuit breakers before each test."""
        circuit_breaker_manager.reset_all()
        yield
        circuit_breaker_manager.reset_all()
    
    def test_latency_under_threshold(self):
        ...
    
    def test_timeout_fails_fast(self):
        ...
    
    def test_circuit_opens_on_errors(self):
        ...
    
    def test_circuit_recovers(self):
        ...
    
    def test_retry_with_backoff(self):
        ...
    
    def test_fallback_on_failure(self):
        ...

Run in CI on a schedule or before major releases:

# CI: run resilience tests nightly
nightly-resilience:
  script: pytest tests/resilience/ -v --tb=short
  only:
    - schedules

Summary

Fault injection gives you confidence that your microservices fail in the ways you expect:

  • Latency injection — verify timeout handling and circuit breaker thresholds
  • Error injection — test fallbacks, retries, and circuit breaker behavior
  • Resource exhaustion — check behavior under connection pool and memory pressure
  • Network partitions — verify split-brain handling and recovery
  • Dependency failures — confirm non-critical paths don't break critical ones
  • Observability verification — confirm monitoring catches injected failures

Start with the failures that have happened before — those runbooks and postmortems tell you exactly what to inject. Then expand to the failures you fear most. The goal is that when the next failure happens, it's one you've already tested.

Read more

Start now free