Add /burn endpoint for CPU load simulation

Author: emyller

src/common/core/urls.py

@@ -11,6 +11,7 @@
     # Aptible health checks must be on /healthcheck and cannot redirect
     # see https://www.aptible.com/docs/core-concepts/apps/connecting-to-apps/app-endpoints/https-endpoints/health-checks
     re_path(r"^healthcheck", include("health_check.urls", namespace="health-aptible")),
+    path("burn/", views.burn),
 ]
 
 if settings.PROMETHEUS_ENABLED:

src/common/core/views.py

@@ -1,4 +1,5 @@
 import logging
+import time
 
 import prometheus_client
 from django.http import HttpResponse, JsonResponse
@@ -25,3 +26,17 @@ def metrics(request: Request) -> HttpResponse:
         metrics_page,
         content_type=prometheus_client.CONTENT_TYPE_LATEST,
     )
+
+
+def burn(request: Request) -> HttpResponse:
+    """
+    Burn CPU for a specified duration to simulate load.
+
+    Usage: GET /burn?s=10 (burns CPU for 10 seconds)
+    """
+    seconds = int(request.GET.get("s", 1))
+    end_time = time.monotonic() + seconds
+    while time.monotonic() < end_time:
+        pass  # Busy wait
+
+    return HttpResponse("waited")

tests/integration/core/test_views.py

@@ -1,5 +1,6 @@
 import prometheus_client
 import pytest
+from pytest_mock import MockerFixture
 from rest_framework.test import APIClient
 
 from common.test_tools import SnapshotFixture
@@ -28,3 +29,33 @@ def test_metrics__return_expected(
     # Assert
     assert response.status_code == 200
     assert response.content.decode() == snapshot()
+
+
+def test_burn__default_duration__returns_expected(
+    client: APIClient,
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mocker.patch("common.core.views.time.monotonic", side_effect=[0, 1])
+
+    # When
+    response = client.get("/burn/")
+
+    # Then
+    assert response.status_code == 200
+    assert response.content.decode() == "waited"
+
+
+def test_burn__custom_duration__returns_expected(
+    client: APIClient,
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mocker.patch("common.core.views.time.monotonic", side_effect=[0, 3])
+
+    # When
+    response = client.get("/burn/?s=3")
+
+    # Then
+    assert response.status_code == 200
+    assert response.content.decode() == "waited"