feat(process_label): enhance uWSGI process role detection

- Added a new function `_is_uwsgi_worker` to accurately identify when the application is running within a uWSGI worker.
- Updated `get_process_role` to label processes as "uwsgi" based on the new function or command-line arguments, improving process role accuracy.
- Expanded unit tests to cover various scenarios for uWSGI role detection, ensuring robust validation of the new logic.
This commit is contained in:
SergeantPanda 2026-06-16 16:21:22 -05:00
parent 5d424adbe8
commit a7b4db83fa
3 changed files with 24 additions and 5 deletions

View file

@ -6,6 +6,15 @@ import os
import sys
def _is_uwsgi_worker() -> bool:
"""True when running inside a uWSGI worker (not master or a non-uWSGI process)."""
try:
import uwsgi
except ImportError:
return False
return uwsgi.worker_id() > 0
def get_process_role(argv: list[str] | None = None) -> str:
argv = argv if argv is not None else sys.argv
argv0 = os.path.basename(argv[0]) if argv else ""
@ -25,7 +34,7 @@ def get_process_role(argv: list[str] | None = None) -> str:
return "daphne"
if argv0 == "manage.py" and len(argv) > 1:
return f"manage-{argv[1]}"
if os.environ.get("GEVENT_SUPPORT"):
if _is_uwsgi_worker() or argv0 == "uwsgi":
return "uwsgi"
return "django"

View file

@ -70,7 +70,6 @@ py-autoreload = 1
honour-stdin = true
# Environment variables
env = GEVENT_SUPPORT=True
env = PYTHONPATH=/app
env = PYTHONUNBUFFERED=1
env = PYDEVD_DISABLE_FILE_VALIDATION=1

View file

@ -12,7 +12,18 @@ class ProcessLabelTests(SimpleTestCase):
)
self.assertEqual(role, "celery-worker")
def test_uwsgi_labeled_when_gevent_support(self):
with patch.dict("os.environ", {"GEVENT_SUPPORT": "True"}):
role = get_process_role(["uwsgi"])
def test_uwsgi_labeled_from_argv(self):
role = get_process_role(["/dispatcharrpy/bin/uwsgi", "--ini", "/app/docker/uwsgi.ini"])
self.assertEqual(role, "uwsgi")
def test_uwsgi_labeled_when_worker_module_present(self):
fake_uwsgi = type("uwsgi", (), {"worker_id": lambda: 2})()
with patch.dict("sys.modules", {"uwsgi": fake_uwsgi}):
role = get_process_role(["/dispatcharrpy/bin/python", "-c", "pass"])
self.assertEqual(role, "uwsgi")
def test_uwsgi_master_not_labeled_as_uwsgi(self):
fake_uwsgi = type("uwsgi", (), {"worker_id": lambda: 0})()
with patch.dict("sys.modules", {"uwsgi": fake_uwsgi}):
role = get_process_role(["/dispatcharrpy/bin/python", "-c", "pass"])
self.assertEqual(role, "django")