stress-testing

name	stress-testing
description	Test system behavior under extreme load conditions to identify breaking points, capacity limits, and failure modes. Use for stress test, capacity testing, breaking point analysis, spike test, and system limits validation.

Stress Testing

Overview

Stress testing pushes systems beyond normal operating capacity to identify breaking points, failure modes, and recovery behavior. It validates system stability under extreme conditions and helps determine maximum capacity before degradation or failure.

When to Use

• Finding system capacity limits
• Identifying breaking points
• Testing auto-scaling behavior
• Validating error handling under load
• Testing recovery after failures
• Planning capacity requirements
• Verifying graceful degradation
• Testing spike traffic handling

Test Types

• Stress Test: Gradually increase load until failure
• Spike Test: Sudden large increase in load
• Soak Test: Sustained high load over extended period
• Capacity Test: Find maximum sustainable load
• Volume Test: Large data volumes
• Scalability Test: Performance at different scales

Instructions

1. k6 Stress Testing

// stress-test.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate } from 'k6/metrics';

const errorRate = new Rate('errors');

export const options = {
  stages: [
    // Stress testing: Progressive load increase
    { duration: '2m', target: 100 },    // Normal load
    { duration: '5m', target: 100 },    // Sustain normal
    { duration: '2m', target: 200 },    // Above normal
    { duration: '5m', target: 200 },    // Sustain above normal
    { duration: '2m', target: 300 },    // Breaking point approaching
    { duration: '5m', target: 300 },    // Sustain high load
    { duration: '2m', target: 400 },    // Beyond capacity
    { duration: '5m', target: 400 },    // System under stress
    { duration: '5m', target: 0 },      // Gradual recovery
  ],
  thresholds: {
    http_req_duration: ['p(99)<1000'],  // 99% under 1s during stress
    http_req_failed: ['rate<0.05'],     // Allow 5% error rate under stress
    errors: ['rate<0.1'],
  },
};

const BASE_URL = __ENV.BASE_URL || 'http://localhost:3000';

export function setup() {
  // Prepare test data
  const res = http.post(`${BASE_URL}/api/auth/login`, {
    email: '[e-mail]',
    password: 'test123',
  });

  return { token: res.json('token') };
}

export default function (data) {
  const headers = {
    Authorization: `Bearer ${data.token}`,
    'Content-Type': 'application/json',
  };

  // Heavy database query
  const productsRes = http.get(
    `${BASE_URL}/api/products?page=1&limit=100`,
    { headers }
  );

  const productsCheck = check(productsRes, {
    'products loaded': (r) => r.status === 200,
    'has products': (r) => r.json('products').length > 0,
  });

  if (!productsCheck) {
    errorRate.add(1);
    console.error(`Products failed: ${productsRes.status} ${productsRes.body}`);
  }

  sleep(1);

  // Write operation - stress database
  const orderPayload = JSON.stringify({
    items: [
      { productId: Math.floor(Math.random() * 100), quantity: 2 },
    ],
  });

  const orderRes = http.post(`${BASE_URL}/api/orders`, orderPayload, {
    headers,
  });

  const orderCheck = check(orderRes, {
    'order created': (r) => r.status === 201 || r.status === 503,
    'response within 5s': (r) => r.timings.duration < 5000,
  });

  if (!orderCheck) {
    errorRate.add(1);
  }

  // Monitor degradation
  if (orderRes.status === 503) {
    console.log('Service unavailable - system at capacity');
  }

  sleep(1);
}

export function teardown(data) {
  // Log final metrics
  console.log('Stress test completed');
}

2. Spike Testing

// spike-test.js
import http from 'k6/http';
import { check } from 'k6';

export const options = {
  stages: [
    { duration: '30s', target: 10 },     // Normal baseline
    { duration: '1m', target: 10 },      // Stable baseline
    { duration: '10s', target: 1000 },   // SPIKE! 100x increase
    { duration: '3m', target: 1000 },    // Maintain spike
    { duration: '10s', target: 10 },     // Drop back
    { duration: '3m', target: 10 },      // Recovery period
  ],
  thresholds: {
    http_req_duration: ['p(95)<5000'],   // Allow degradation during spike
    http_req_failed: ['rate<0.1'],       // Allow 10% errors during spike
  },
};

export default function () {
  const res = http.get();

  (res, {
    :  r. ===  || r. === ,
    :  r.. > ,
  });
}

3. Soak/Endurance Testing

# soak_test.py
import asyncio
import aiohttp
import time
from datetime import datetime, timedelta
import psutil
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SoakTest:
    """Run sustained load test to detect memory leaks and degradation."""

    def __init__(self, url, duration_hours=4, requests_per_second=50):
        self.url = url
        self.duration = timedelta(hours=duration_hours)
        self.rps = requests_per_second
        self.metrics = {
            'requests': 0,
            'errors': 0,
            'response_times': [],
            'memory_usage': [],
        }

    async def make_request(self, session):
        """Make single request and record metrics."""
        start = time.time()
        try:
            async with session.get(self.url) as response:
                await response.read()
                duration = time.time() - start

                self.metrics['requests'] += 1
                self.metrics['response_times'].append(duration)

                if response.status >= :
                    .metrics[] += 
                    logger.warning()

         Exception  e:
            .metrics[] += 
            logger.error()

      ():
        
         .running:
             .make_request(session)
             asyncio.sleep( / .rps)

     ():
        
        process = psutil.Process()
         {
            : process.memory_info().rss /  / ,
            : process.cpu_percent(),
            : datetime.now(),
        }

      ():
        
        start_time = datetime.now()
        end_time = start_time + .duration
        .running = 

        logger.info()
        logger.info()

          aiohttp.ClientSession()  session:
            
            workers = [
                asyncio.create_task(.worker(session))
                 _  ()  
            ]

            
             datetime.now() < end_time:
                 asyncio.sleep()  

                resources = .monitor_resources()
                .metrics[].append(resources)

                
                elapsed = (datetime.now() - start_time).total_seconds()
                error_rate = .metrics[] / (.metrics[], )
                avg_response = (.metrics[][-:]) / 

                logger.info(
                    
                    
                    
                    
                    
                )

                
                 (.metrics[]) > :
                    initial_mem = .metrics[][][]
                    current_mem = resources[]
                    growth = current_mem - initial_mem

                     growth > :  
                        logger.warning()

            
            .running = 
             asyncio.gather(*workers, return_exceptions=)

        .report()

     ():
        
        total_requests = .metrics[]
        error_rate = .metrics[] / total_requests  total_requests >   
        response_times = .metrics[]

        ( + *)
        ()
        (*)
        ()
        ()
        ()
        ()
        ()
        ()
        ()
        ()

        
         .metrics[]:
            initial_mem = .metrics[][][]
            final_mem = .metrics[][-][]
            growth = final_mem - initial_mem

            ()
            ()
            ()
            ()

             growth > :
                ()

        (*)


 __name__ == :
    test = SoakTest(
        url=,
        duration_hours=,
        requests_per_second=
    )
    asyncio.run(test.run())

4. JMeter Stress Test

<!-- stress-test.jmx -->
<jmeterTestPlan>
  <ThreadGroup testname="Stress Test Thread Group">
    <!-- Ultimate Thread Group for advanced load patterns -->
    <elementProp name="ThreadGroup.main_controller">
      <!-- Stage 1: Ramp up to 100 users -->
      <collectionProp name="ultimatethreadgroupdata">
        <stringProp>100</stringProp>  <!-- Users -->
        <stringProp>60</stringProp>   <!-- Ramp-up (sec) -->
        <stringProp>300</stringProp>  <!-- Duration (sec) -->
      </collectionProp>

      <!-- Stage 2: Ramp up to 500 users -->
      <collectionProp name="ultimatethreadgroupdata">
        <stringProp>500</stringProp>
        <stringProp>120</stringProp>
        <stringProp>600</stringProp>
      </collectionProp>

      <!-- Stage 3: Ramp up to 1000 users (stress) -->
      
        1000
        180
        600
      
    

    
      api.example.com
      /api/search?q=stress
      GET
    

    
    
      Assertion.response_code
      8
      200|503
    
  

5. Auto-Scaling Validation

// test-autoscaling.ts
import { test, expect } from '@playwright/test';
import axios from 'axios';

test.describe('Auto-scaling Stress Test', () => {
  test('system should scale up under load', async () => {
    const baseUrl = 'http://api.example.com';
    const cloudwatch = new AWS.CloudWatch();

    // Initial instance count
    const initialInstances = await getInstanceCount();
    console.log(`Initial instances: ${initialInstances}`);

    // Generate high load
    const requests = [];
    for (let i = 0; i < 1000; i++) {
      requests.push(
        axios.get(`${baseUrl}/api/heavy-operation`)
          .catch(err => ({ error: err.message }))
      );
    }

    // Wait for auto-scaling trigger
    await Promise.all(requests);
    await new Promise( (resolve, )); 

    
     scaledInstances =  ();
    .();

    (scaledInstances).(initialInstances);

    
     cpuMetrics =  cloudwatch.({
      : ,
      : ,
      
    }).();

    (cpuMetrics..( d. > )).();
  });
});

6. Breaking Point Analysis

# find_breaking_point.py
import requests
import threading
import time
from collections import defaultdict

class BreakingPointTest:
    """Find system breaking point by gradually increasing load."""

    def __init__(self, url):
        self.url = url
        self.results = defaultdict(lambda: {'success': 0, 'errors': 0, 'times': []})
        self.running = True

    def worker(self, vusers):
        """Worker thread that makes requests."""
        while self.running:
            start = time.time()
            try:
                response = requests.get(self.url, timeout=10)
                duration = time.time() - start

                if response.status_code == 200:
                    self.results[vusers]['success'] += 1
                    self.results[vusers]['times'].append(duration)
                else:
                    self.results[vusers]['errors'] += 1

            except Exception as e:
                self.results[vusers][] += 

            time.sleep()

     ():
        
        ()

        threads = []
         _  (vusers):
            t = threading.Thread(target=.worker, args=(vusers,))
            t.start()
            threads.append(t)

        time.sleep(duration)

        .running = 
         t  threads:
            t.join()

        .running = 

        
        stats = .results[vusers]
        total = stats[] + stats[]
        error_rate = stats[] / total  total >   
        avg_time = (stats[]) / (stats[])  stats[]  

        ()
        ()
        ()
        ()
        ()

        
        is_breaking = error_rate >   avg_time > 

          is_breaking

     ():
        
        min_users = 
        max_users = 
        breaking_point = 

         min_users < max_users:
            mid = (min_users + max_users) // 

             .test_load_level(mid):
                
                min_users = mid + 
            :
                
                breaking_point = mid
                max_users = mid - 

        ()
        ()
        ()

         breaking_point


test = BreakingPointTest()
test.find_breaking_point()

Metrics to Monitor

Application Metrics

• Response times (P50, P95, P99, Max)
• Error rates and types
• Throughput (req/s)
• Queue depths
• Circuit breaker trips

System Metrics

• CPU utilization
• Memory usage and leaks
• Disk I/O
• Network bandwidth
• Thread/connection pools

Database Metrics

• Query execution times
• Connection pool usage
• Lock contention
• Cache hit rates
• Replication lag

Best Practices

✅ DO

• Test in production-like environment
• Monitor all system resources
• Gradually increase load to find limits
• Test recovery after stress
• Document breaking points
• Test auto-scaling behavior
• Plan for graceful degradation
• Monitor for memory leaks

❌ DON'T

• Test in production without safeguards
• Skip recovery testing
• Ignore warning signs (CPU, memory)
• Test only success scenarios
• Assume linear scalability
• Forget database capacity
• Skip monitoring third-party dependencies
• Test without proper cleanup

Tools

• Load Generation: k6, JMeter, Gatling, Locust, Artillery
• Monitoring: Prometheus, Grafana, DataDog, New Relic
• Cloud Metrics: CloudWatch, Azure Monitor, GCP Monitoring
• Profiling: py-spy, async-profiler, clinic.js

Examples

See also: performance-testing, continuous-testing, api-versioning-strategy for comprehensive system testing.