feat(middleware): Log request context with large RSS growth

src/sentry/middleware/request_memory.py

@@ -11,6 +11,15 @@
 `middleware.request_rss_growth` (megabytes), tagged with the URL name and, for
 internal RPC calls, the RPC method.
 
+The log line adds context for finding what made the request large: duration,
+status, request and response sizes, user agent, webhook event type, the
+organization from the URL, and the values of a few paging and dataset query
+parameters (`_LOGGED_QUERY_PARAMS`; for others, such as a search query, only
+the length). It also records how many requests the worker had served and its
+uptime, since a new worker's first requests grow RSS while it warms up;
+requests in a worker's first `_WARMUP_REQUESTS` requests or `_WARMUP_SECONDS`
+seconds are tagged `warmup`, on the metric and the log line.
+
 RSS belongs to the whole process, so a request that overlaps others on the
 same worker can be blamed for their allocations too. The `concurrent` tag and
 the `overlapping_requests` log field mark those.
@@ -20,19 +29,21 @@
 `SENTRY_REQUEST_RSS_THRESHOLD_MB` (default 100). Both are read when the
 middleware is created, at worker start. When off, Django drops the middleware
 from the stack, so it costs nothing per request. When on, it costs two short
-reads of /proc/self/statm per request.
+reads of /proc/self/statm and two clock reads per request.
 """
 
 from __future__ import annotations
 
 import logging
 import os
 import threading
+import time
 from collections.abc import Callable
+from typing import Any
 
 from django.core.exceptions import MiddlewareNotUsed
 from django.http.request import HttpRequest
-from django.http.response import HttpResponseBase
+from django.http.response import HttpResponse, HttpResponseBase
 
 from sentry.utils import metrics
 
@@ -45,6 +56,27 @@
 _ENABLED_VALUES = frozenset(("1", "true", "t", "yes", "y", "on"))
 _STATM_PATH = "/proc/self/statm"
 _MB = 1024 * 1024
+# Query parameters whose values are logged. Anything else, such as a search
+# query, could hold user input, so only its presence or length is logged.
+_LOGGED_QUERY_PARAMS = (
+    "per_page",
+    "limit",
+    "full",
+    "dataset",
+    "sort",
+    "expand",
+    "collapse",
+    "statsPeriod",
+    "referrer",
+)
+_MAX_LOGGED_VALUE_LENGTH = 200
+# A worker's first requests grow RSS while it imports modules and fills
+# caches. Requests within either limit are tagged `warmup` so they can be
+# told apart from real offenders.
+_WARMUP_REQUESTS = 10
+_WARMUP_SECONDS = 60
+# Headers that name the event type of an incoming integration webhook.
+_WEBHOOK_EVENT_HEADERS = ("X-GitHub-Event", "X-Gitlab-Event", "X-Event-Key")
 
 
 def is_enabled() -> bool:
@@ -67,6 +99,43 @@ def read_rss_bytes() -> int:
     return resident_pages * os.sysconf("SC_PAGE_SIZE")
 
 
+def _query_context(request: HttpRequest) -> dict[str, Any]:
+    # Flat keys, so each one is a queryable log attribute.
+    params = request.GET
+    context: dict[str, Any] = {
+        f"param_{name}": params.get(name, "")[:_MAX_LOGGED_VALUE_LENGTH]
+        for name in _LOGGED_QUERY_PARAMS
+        if name in params
+    }
+    if "field" in params:
+        context["field_count"] = len(params.getlist("field"))
+    if "query" in params:
+        context["query_length"] = len(params.get("query", ""))
+    return context
+
+
+def _webhook_event(request: HttpRequest) -> str | None:
+    for header in _WEBHOOK_EVENT_HEADERS:
+        value = request.headers.get(header)
+        if value:
+            return value[:_MAX_LOGGED_VALUE_LENGTH]
+    return None
+
+
+def _request_bytes(request: HttpRequest) -> int | None:
+    try:
+        return int(request.META.get("CONTENT_LENGTH") or "")
+    except ValueError:
+        return None
+
+
+def _response_bytes(response: HttpResponseBase | None) -> int | None:
+    # Streaming responses haven't produced their body yet.
+    if isinstance(response, HttpResponse):
+        return len(response.content)
+    return None
+
+
 def _rpc_method(request: HttpRequest) -> str | None:
     match = request.resolver_match
     if match is None:
@@ -97,6 +166,7 @@ def __init__(
         self._lock = threading.Lock()
         self._inflight = 0
         self._started = 0
+        self._created_at = time.monotonic()
 
     def _try_read_rss(self) -> int | None:
         # A failed read must never fail the request or mask its exception.
@@ -112,27 +182,58 @@ def __call__(self, request: HttpRequest) -> HttpResponseBase:
             inflight_at_start = self._inflight
             started_at_start = self._started
         before = self._try_read_rss()
+        start = time.monotonic()
+        response: HttpResponseBase | None = None
         try:
-            return self.get_response(request)
+            response = self.get_response(request)
+            return response
         finally:
+            duration = time.monotonic() - start
             after = self._try_read_rss()
             with self._lock:
                 self._inflight -= 1
                 # Requests already running when this one started, plus any
                 # that started while it ran.
                 overlapping = inflight_at_start - 1 + self._started - started_at_start
             if before is not None and after is not None and after - before >= self.threshold:
-                self._report(request, before, after, overlapping)
+                # Reporting runs in `finally`, so it must never fail the
+                # request or mask the view's exception.
+                try:
+                    self._report(
+                        request,
+                        response,
+                        before,
+                        after,
+                        overlapping,
+                        duration=duration,
+                        worker_request_number=started_at_start,
+                        worker_uptime=time.monotonic() - self._created_at,
+                    )
+                except Exception:
+                    logger.exception("request.rss_growth.report_failed")
 
-    def _report(self, request: HttpRequest, before: int, after: int, overlapping: int) -> None:
+    def _report(
+        self,
+        request: HttpRequest,
+        response: HttpResponseBase | None,
+        before: int,
+        after: int,
+        overlapping: int,
+        *,
+        duration: float,
+        worker_request_number: int,
+        worker_uptime: float,
+    ) -> None:
         match = request.resolver_match
         url_name = (match.url_name if match else None) or "unknown"
         rpc_method = _rpc_method(request)
         growth_mb = (after - before) // _MB
+        warmup = worker_request_number <= _WARMUP_REQUESTS or worker_uptime < _WARMUP_SECONDS
         tags = {
             "url_name": url_name,
             "rpc_method": rpc_method or "none",
             "concurrent": "true" if overlapping > 0 else "false",
+            "warmup": "true" if warmup else "false",
         }
         # These are rare, so record every one instead of the default sample rate.
         metrics.distribution(
@@ -142,6 +243,7 @@ def _report(self, request: HttpRequest, before: int, after: int, overlapping: in
             sample_rate=1.0,
             unit="megabyte",
         )
+        organization = match.kwargs.get("organization_id_or_slug") if match else None
         logger.info(
             "request.rss_growth",
             extra={
@@ -152,5 +254,16 @@ def _report(self, request: HttpRequest, before: int, after: int, overlapping: in
                 "rss_after_mb": after // _MB,
                 "growth_mb": growth_mb,
                 "overlapping_requests": overlapping,
+                "duration_ms": int(duration * 1000),
+                "status_code": response.status_code if response is not None else None,
+                "request_bytes": _request_bytes(request),
+                "response_bytes": _response_bytes(response),
+                "user_agent": request.headers.get("User-Agent", "")[:_MAX_LOGGED_VALUE_LENGTH],
+                "webhook_event": _webhook_event(request),
+                "organization": organization,
+                **_query_context(request),
+                "worker_request_number": worker_request_number,
+                "worker_uptime_s": int(worker_uptime),
+                "warmup": warmup,
             },
         )

tests/sentry/middleware/test_request_memory.py

@@ -4,7 +4,7 @@
 
 import pytest
 from django.core.exceptions import MiddlewareNotUsed
-from django.http import HttpRequest, HttpResponse
+from django.http import HttpRequest, HttpResponse, StreamingHttpResponse
 from django.http.response import HttpResponseBase
 from django.test import RequestFactory
 from django.urls import ResolverMatch
@@ -111,6 +111,7 @@ def test_large_rpc_growth_is_reported(distribution: mock.MagicMock, enabled: Non
             "url_name": "sentry-api-0-rpc-service",
             "rpc_method": "organization.get_organization",
             "concurrent": "false",
+            "warmup": "true",
         },
         sample_rate=1.0,
         unit="megabyte",
@@ -123,6 +124,124 @@ def test_large_rpc_growth_is_reported(distribution: mock.MagicMock, enabled: Non
     assert extra["rss_after_mb"] == 1100
     assert extra["rpc_method"] == "organization.get_organization"
     assert extra["overlapping_requests"] == 0
+    assert extra["status_code"] == 200
+    assert extra["response_bytes"] == len(b"ok")
+    assert extra["request_bytes"] == int(request.META["CONTENT_LENGTH"])
+    assert extra["duration_ms"] >= 0
+    assert extra["worker_request_number"] == 1
+    assert extra["worker_uptime_s"] >= 0
+    assert extra["webhook_event"] is None
+    assert extra["organization"] is None
+    assert not any(key.startswith("param_") for key in extra)
+    assert "field_count" not in extra
+    assert "query_length" not in extra
+
+
+@mock.patch("sentry.middleware.request_memory.metrics.distribution")
+def test_logs_request_context(distribution: mock.MagicMock, enabled: None) -> None:
+    middleware = _middleware([400 * MB, 600 * MB])
+    request = RequestFactory().get(
+        "/api/0/organizations/org-slug/issues/1/events/",
+        {
+            "per_page": "100",
+            "full": "true",
+            "query": "user.email:user@example.com",
+            "field": ["title", "timestamp"],
+            "cursor": "0:100:0",
+        },
+        HTTP_USER_AGENT="sentry-mcp/1.0",
+        HTTP_X_GITHUB_EVENT="push",
+    )
+    request.resolver_match = ResolverMatch(
+        lambda r: None,
+        (),
+        {"organization_id_or_slug": "org-slug", "issue_id": "1"},
+        url_name="sentry-api-0-organization-group-group-events",
+    )
+    with mock.patch("sentry.middleware.request_memory.logger") as logger:
+        middleware(request)
+
+    extra = logger.info.call_args.kwargs["extra"]
+    # Allowlisted parameters keep their values; the search query only its length.
+    assert extra["param_per_page"] == "100"
+    assert extra["param_full"] == "true"
+    assert extra["field_count"] == 2
+    assert extra["query_length"] == len("user.email:user@example.com")
+    assert "param_query" not in extra
+    assert "param_cursor" not in extra
+    assert "user@example.com" not in str(extra)
+    assert extra["user_agent"] == "sentry-mcp/1.0"
+    assert extra["webhook_event"] == "push"
+    assert extra["organization"] == "org-slug"
+
+
+@mock.patch("sentry.middleware.request_memory.metrics.distribution")
+def test_failed_report_does_not_fail_the_request(
+    distribution: mock.MagicMock, enabled: None
+) -> None:
+    middleware = _middleware([400 * MB, 600 * MB])
+    with (
+        mock.patch("sentry.middleware.request_memory.logger") as logger,
+        mock.patch(
+            "sentry.middleware.request_memory._query_context", side_effect=RuntimeError("bad")
+        ),
+    ):
+        response = middleware(_request("sentry-api-0-organization-details"))
+    assert response.status_code == 200
+    logger.exception.assert_called_once_with("request.rss_growth.report_failed")
+
+
+@mock.patch("sentry.middleware.request_memory.metrics.distribution")
+def test_counts_requests_served_by_the_worker(distribution: mock.MagicMock, enabled: None) -> None:
+    middleware = _middleware([400 * MB, 410 * MB, 410 * MB, 600 * MB])
+    with mock.patch("sentry.middleware.request_memory.logger") as logger:
+        middleware(_request("first"))
+        middleware(_request("second"))
+    logger.info.assert_called_once()
+    assert logger.info.call_args.kwargs["extra"]["worker_request_number"] == 2
+
+
+def _report_warmup(middleware: RequestMemoryMiddleware) -> tuple[str, bool]:
+    with (
+        mock.patch("sentry.middleware.request_memory.metrics.distribution") as distribution,
+        mock.patch("sentry.middleware.request_memory.logger") as logger,
+    ):
+        middleware(_request("sentry-api-0-organization-details"))
+    return (
+        distribution.call_args.kwargs["tags"]["warmup"],
+        logger.info.call_args.kwargs["extra"]["warmup"],
+    )
+
+
+def test_warm_worker_is_not_warmup(enabled: None) -> None:
+    middleware = _middleware([400 * MB, 600 * MB])
+    # Past both limits: many requests served and well over a minute old.
+    middleware._started = 50
+    middleware._created_at -= 120
+    assert _report_warmup(middleware) == ("false", False)
+
+
+def test_early_request_is_warmup_even_on_an_old_worker(enabled: None) -> None:
+    middleware = _middleware([400 * MB, 600 * MB])
+    middleware._created_at -= 120
+    assert _report_warmup(middleware) == ("true", True)
+
+
+def test_young_worker_is_warmup_even_after_many_requests(enabled: None) -> None:
+    middleware = _middleware([400 * MB, 600 * MB])
+    middleware._started = 50
+    assert _report_warmup(middleware) == ("true", True)
+
+
+@mock.patch("sentry.middleware.request_memory.metrics.distribution")
+def test_streaming_response_has_no_size(distribution: mock.MagicMock, enabled: None) -> None:
+    values = _fake_rss([400 * MB, 600 * MB])
+    middleware = RequestMemoryMiddleware(
+        lambda request: StreamingHttpResponse(iter([b"chunk"])), read_rss=lambda: next(values)
+    )
+    with mock.patch("sentry.middleware.request_memory.logger") as logger:
+        middleware(_request("sentry-api-0-organization-details"))
+    assert logger.info.call_args.kwargs["extra"]["response_bytes"] is None
 
 
 @mock.patch("sentry.middleware.request_memory.metrics.distribution")
@@ -143,10 +262,16 @@ def boom(request: HttpRequest) -> HttpResponse:
         raise RuntimeError("boom")
 
     middleware = RequestMemoryMiddleware(boom, read_rss=lambda: next(values))
-    with mock.patch("sentry.middleware.request_memory.logger"), pytest.raises(RuntimeError):
+    with (
+        mock.patch("sentry.middleware.request_memory.logger") as logger,
+        pytest.raises(RuntimeError),
+    ):
         middleware(_request("sentry-api-0-organization-details"))
     distribution.assert_called_once()
     assert distribution.call_args.args[1] == 400
+    extra = logger.info.call_args.kwargs["extra"]
+    assert extra["status_code"] is None
+    assert extra["response_bytes"] is None
 
 
 @mock.patch("sentry.middleware.request_memory.metrics.distribution")