k6 vs Locust: Which Load Testing Tool Should Your Team Use?

// api-load-test.js
import http from 'k6/http';
import { check, sleep, group } from 'k6';
import { Rate, Trend } from 'k6/metrics';

// Custom metrics
const checkoutErrors = new Rate('checkout_errors');
const checkoutDuration = new Trend('checkout_duration', true);

export const options = {
  stages: [
    { duration: '2m', target: 50 },   // Ramp up to 50 users
    { duration: '5m', target: 50 },   // Hold at 50 users
    { duration: '2m', target: 100 },  // Ramp up to 100 users
    { duration: '5m', target: 100 },  // Hold at 100 users
    { duration: '2m', target: 0 },    // Ramp down
  ],
  thresholds: {
    http_req_duration: ['p(95)<500', 'p(99)<2000'],  // 95th pct < 500ms
    http_req_failed: ['rate<0.01'],                   // Error rate < 1%
    checkout_errors: ['rate<0.005'],                  // Checkout errors < 0.5%
  },
};

const BASE_URL = __ENV.BASE_URL || 'https://api.staging.example.com';

// Shared setup: runs once before all VUs
export function setup() {
  // Pre-create test users (one per VU would be created in the test)
  return { startTime: Date.now() };
}

export default function (data) {
  let authToken;

  // Authentication group
  group('authentication', () => {
    const loginRes = http.post(`${BASE_URL}/auth/login`, JSON.stringify({
      email: `testuser+${__VU}@example.com`,
      password: 'TestPassword123!',
    }), {
      headers: { 'Content-Type': 'application/json' },
    });

    check(loginRes, {
      'login status is 200': (r) => r.status === 200,
      'login returns token': (r) => r.json('token') !== undefined,
    });

    if (loginRes.status === 200) {
      authToken = loginRes.json('token');
    }
  });

  if (!authToken) {
    return;  // Skip remaining steps if auth failed
  }

  const headers = {
    'Content-Type': 'application/json',
    'Authorization': `Bearer ${authToken}`,
  };

  // Browse products group
  group('browse_products', () => {
    const productsRes = http.get(`${BASE_URL}/products?page=1&per_page=20`, { headers });

    check(productsRes, {
      'products status is 200': (r) => r.status === 200,
      'products returns items': (r) => r.json('data').length > 0,
    });

    sleep(2);  // Simulate reading product list (2-second think time)

    // View a product detail (simulate clicking on first product)
    const products = productsRes.json('data');
    if (products && products.length > 0) {
      const productId = products[Math.floor(Math.random() * products.length)].id;
      const detailRes = http.get(`${BASE_URL}/products/${productId}`, { headers });

      check(detailRes, {
        'product detail status is 200': (r) => r.status === 200,
      });

      sleep(3);  // Simulate reading product detail
    }
  });

  // Checkout group
  group('checkout', () => {
    const checkoutStart = Date.now();

    // Add to cart
    const cartRes = http.post(`${BASE_URL}/cart/items`, JSON.stringify({
      product_id: `prod_${Math.floor(Math.random() * 1000) + 1}`,
      quantity: 1,
    }), { headers });

    const cartOk = check(cartRes, {
      'add to cart status is 201': (r) => r.status === 201,
    });

    if (cartOk) {
      sleep(1);

      // Complete checkout
      const orderRes = http.post(`${BASE_URL}/orders`, JSON.stringify({
        cart_id: cartRes.json('cart_id'),
        payment_method: 'test_card',
      }), { headers });

      const checkoutOk = check(orderRes, {
        'checkout status is 201': (r) => r.status === 201,
        'checkout returns order id': (r) => r.json('order_id') !== undefined,
      });

      checkoutErrors.add(!checkoutOk);
      checkoutDuration.add(Date.now() - checkoutStart);
    }
  });

  sleep(1 + Math.random() * 2);  // Random think time between 1-3 seconds
}

export function teardown(data) {
  console.log(`Test completed. Duration: ${Date.now() - data.startTime}ms`);
}

# locustfile.py
import random
import time
from locust import HttpUser, TaskSet, task, between, events
from locust.runners import MasterRunner

# Custom metric tracking
checkout_errors = 0
checkout_count = 0


class UserJourney(TaskSet):
    auth_token = None

    def on_start(self):
        """Called when a user starts. Used for authentication."""
        with self.client.post(
            "/auth/login",
            json={
                "email": f"testuser+{id(self)}@example.com",
                "password": "TestPassword123!",
            },
            catch_response=True,
            name="POST /auth/login",
        ) as response:
            if response.status_code == 200:
                self.auth_token = response.json().get("token")
            else:
                response.failure(f"Login failed: {response.status_code}")

    @property
    def auth_headers(self):
        return {
            "Authorization": f"Bearer {self.auth_token}",
            "Content-Type": "application/json",
        }

    @task(10)  # Runs 10x more frequently than checkout
    def browse_products(self):
        """Simulate browsing the product catalog."""
        with self.client.get(
            "/products?page=1&per_page=20",
            headers=self.auth_headers,
            name="GET /products",
            catch_response=True,
        ) as response:
            if response.status_code != 200:
                response.failure(f"Products request failed: {response.status_code}")
                return

            products = response.json().get("data", [])
            if not products:
                return

        time.sleep(2)  # Think time: reading product list

        # View a random product detail
        product_id = random.choice(products)["id"]
        self.client.get(
            f"/products/{product_id}",
            headers=self.auth_headers,
            name="GET /products/{id}",
        )
        time.sleep(3)  # Think time: reading product detail

    @task(1)  # Less frequent - checkout is a conversion event
    def checkout_flow(self):
        """Simulate adding to cart and checking out."""
        global checkout_errors, checkout_count

        start_time = time.time()

        # Add to cart
        product_id = f"prod_{random.randint(1, 1000)}"
        with self.client.post(
            "/cart/items",
            json={"product_id": product_id, "quantity": 1},
            headers=self.auth_headers,
            name="POST /cart/items",
            catch_response=True,
        ) as cart_response:
            if cart_response.status_code != 201:
                checkout_errors += 1
                checkout_count += 1
                cart_response.failure(f"Add to cart failed: {cart_response.status_code}")
                return

            cart_id = cart_response.json().get("cart_id")

        time.sleep(1)

        # Complete checkout
        with self.client.post(
            "/orders",
            json={"cart_id": cart_id, "payment_method": "test_card"},
            headers=self.auth_headers,
            name="POST /orders",
            catch_response=True,
        ) as order_response:
            checkout_count += 1
            if order_response.status_code != 201:
                checkout_errors += 1
                order_response.failure(f"Checkout failed: {order_response.status_code}")
            else:
                duration_ms = (time.time() - start_time) * 1000
                # Log custom metric to Locust events
                events.request.fire(
                    request_type="CUSTOM",
                    name="checkout_total_duration",
                    response_time=duration_ms,
                    response_length=0,
                )


class EcommerceUser(HttpUser):
    tasks = [UserJourney]
    wait_time = between(1, 3)  # Random wait between 1 and 3 seconds between task sets

    host = "https://api.staging.example.com"

# Single machine (opens web UI at http://localhost:8089)
locust -f locustfile.py

# Headless mode (CI/CD)
locust -f locustfile.py \
  --headless \
  --users 100 \
  --spawn-rate 10 \
  --run-time 10m \
  --host https://api.staging.example.com