Created per-worker RSS memory tracker via Prometheus resolved #6814

api/app/settings/common.py

@@ -1430,6 +1430,9 @@
 
 PROMETHEUS_ENABLED = env.bool("PROMETHEUS_ENABLED", False)
 
+if PROMETHEUS_ENABLED:
+    MIDDLEWARE.append("core.middleware.worker_rss.WorkerRSSMiddleware")
+
 DOCGEN_MODE = env.bool("DOCGEN_MODE", default=False)
 
 REQUIRE_AUTHENTICATION_FOR_API_DOCS = env.bool(

api/core/middleware/worker_rss.py

@@ -0,0 +1,16 @@
+from django.http import HttpRequest, HttpResponse
+
+from metrics.worker_metrics import update_worker_metrics
+
+
+class WorkerRSSMiddleware:
+    def __init__(self, get_response):  # type: ignore[no-untyped-def]
+        self.get_response = get_response
+
+    def __call__(self, request: HttpRequest) -> HttpResponse:
+        response = self.get_response(request)
+        try:
+            update_worker_metrics()
+        except Exception:
+            pass
+        return response

api/metrics/worker_metrics.py

@@ -0,0 +1,86 @@
+import os
+from pathlib import Path
+from typing import Iterable
+
+import prometheus_client
+
+PROC_SELF_STATUS_PATH = Path("/proc/self/status")
+MAX_RSS_KB_TO_BYTES = 1024
+MAX_RSS_STATUS_FIELD = "VmHWM"
+
+flagsmith_worker_rss_bytes = prometheus_client.Gauge(
+    "flagsmith_worker_rss_bytes",
+    "Maximum RSS (high-water mark) of the worker process in bytes, read from VmHWM in /proc/self/status.",
+    ["pid"],
+    multiprocess_mode="liveall",
+)
+
+
+def update_worker_metrics() -> None:
+    """
+    Update the RSS gauge with the current worker process high-water mark.
+    """
+    current_pid = os.getpid()
+
+    rss_value = get_current_process_max_rss_bytes()
+    if rss_value is not None:
+        flagsmith_worker_rss_bytes.labels(pid=str(current_pid)).set(rss_value)
+
+
+def clear_worker_metrics() -> None:
+    """
+    Clear the RSS memory usage metric for the current worker process.
+    This should be called when a worker process is shutting down to prevent stale metrics.
+    """
+    current_pid = os.getpid()
+    try:
+        flagsmith_worker_rss_bytes.remove(pid=str(current_pid))
+    except (KeyError, ValueError):
+        pass
+
+
+def get_current_process_max_rss_bytes() -> int | None:
+    try:
+        proc_status_lines = PROC_SELF_STATUS_PATH.read_text(
+            encoding="utf-8"
+        ).splitlines()
+    except (FileNotFoundError, OSError, UnicodeDecodeError):
+        return None
+
+    max_rss_kb = _get_proc_status_memory_kb(proc_status_lines, MAX_RSS_STATUS_FIELD)
+    if max_rss_kb is None:
+        return None
+
+    return max_rss_kb * MAX_RSS_KB_TO_BYTES
+
+
+def _get_proc_status_memory_kb(
+    proc_status_lines: Iterable[str],
+    field_name: str,
+) -> int | None:
+    for line in proc_status_lines:
+        name, separator, value = line.strip().partition(":")
+        if separator and name == field_name:
+            return _parse_proc_status_memory_kb(value)
+
+    return None
+
+
+def _parse_proc_status_memory_kb(value: str) -> int | None:
+    parts = value.split()
+    if len(parts) != 2:
+        return None
+
+    memory_kb_text, unit = parts
+    if unit != "kB":
+        return None
+
+    try:
+        memory_kb = int(memory_kb_text)
+    except ValueError:
+        return None
+
+    if memory_kb < 0:
+        return None
+
+    return memory_kb

api/tests/integration/core/test_integration_core_worker_rss_metric.py

@@ -0,0 +1,46 @@
+import os
+
+from django.conf import settings as django_settings
+from django.test import Client, override_settings
+from prometheus_client import REGISTRY, generate_latest
+from pytest_mock import MockerFixture
+
+from metrics.worker_metrics import clear_worker_metrics
+
+
+@override_settings(
+    MIDDLEWARE=[
+        *django_settings.MIDDLEWARE,
+        "core.middleware.worker_rss.WorkerRSSMiddleware",
+    ]
+)
+def test_worker_rss_metric__request_through_middleware__appears_in_prometheus_output(
+    client: Client,
+    mocker: MockerFixture,
+) -> None:
+    # Given - deterministic RSS reading so the test is independent of /proc availability
+    # on macOS/Windows CI runners.
+    expected_rss = 12_345_678
+    mocker.patch(
+        "metrics.worker_metrics.get_current_process_max_rss_bytes",
+        return_value=expected_rss,
+    )
+
+    # When - any cheap, known-reachable endpoint trips the middleware after response.
+    response = client.get("/api/v1/swagger.json", HTTP_ACCEPT="application/json")
+
+    # Then - the response is unaffected by the middleware, and the gauge is exposed
+    # with a sample for the current worker's PID via the Prometheus exposition format.
+    assert response.status_code == 200
+    output = generate_latest(REGISTRY).decode()
+    assert "flagsmith_worker_rss_bytes" in output
+    assert f'pid="{os.getpid()}"' in output
+
+
+def teardown_function(function: object) -> None:
+    # Prevent labelled-child leakage to other tests in the same xdist worker by removing
+    # this PID's sample after each test. Uses the existing module API.
+    try:
+        clear_worker_metrics()
+    except Exception:
+        pass

api/tests/unit/core/middleware/test_unit_core_middleware_worker_rss.py

@@ -0,0 +1,53 @@
+from django.http import HttpResponse
+
+from core.middleware.worker_rss import WorkerRSSMiddleware
+
+
+def test_worker_rss_middleware__any_request__calls_update_after_response(mocker):  # type: ignore[no-untyped-def]
+    # Given
+    call_order = []
+
+    def fake_get_response(request):  # type: ignore[no-untyped-def]
+        call_order.append("handled")
+        return HttpResponse()
+
+    mocker.patch(
+        "core.middleware.worker_rss.update_worker_metrics",
+        side_effect=lambda: call_order.append("updated"),
+    )
+    middleware = WorkerRSSMiddleware(fake_get_response)
+
+    # When
+    middleware(mocker.MagicMock())
+
+    # Then — metric must be updated after the request is handled, not before
+    assert call_order == ["handled", "updated"]
+
+
+def test_worker_rss_middleware__any_request__returns_response_unchanged(mocker):  # type: ignore[no-untyped-def]
+    # Given
+    expected_response = HttpResponse(status=200)
+    mocker.patch("core.middleware.worker_rss.update_worker_metrics")
+    middleware = WorkerRSSMiddleware(lambda _request: expected_response)
+
+    # When
+    result = middleware(mocker.MagicMock())
+
+    # Then
+    assert result is expected_response
+
+
+def test_worker_rss_middleware__update_raises__request_still_completes(mocker):  # type: ignore[no-untyped-def]
+    # Given
+    expected_response = HttpResponse(status=200)
+    mocker.patch(
+        "core.middleware.worker_rss.update_worker_metrics",
+        side_effect=Exception("metric failure"),
+    )
+    middleware = WorkerRSSMiddleware(lambda _request: expected_response)
+
+    # When
+    result = middleware(mocker.MagicMock())
+
+    # Then — exception is swallowed, response still returned
+    assert result is expected_response