diff --git a/src/sentry/middleware/request_memory.py b/src/sentry/middleware/request_memory.py index 35bf7bfa5b22..485c9c1a5526 100644 --- a/src/sentry/middleware/request_memory.py +++ b/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,7 +29,7 @@ `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 @@ -28,11 +37,13 @@ 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,9 +182,13 @@ 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 @@ -122,17 +196,44 @@ def __call__(self, request: HttpRequest) -> HttpResponseBase: # 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, }, ) diff --git a/tests/sentry/middleware/test_request_memory.py b/tests/sentry/middleware/test_request_memory.py index d7930e5ca259..f8dce70f645f 100644 --- a/tests/sentry/middleware/test_request_memory.py +++ b/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")