53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
import logging
|
|
import re
|
|
|
|
_PROBE_PATH = re.compile(
|
|
r"^(?:/live|/ready|/metrics|/healthz|/health[^ ]*|/ping)/?(?:\?|$)"
|
|
)
|
|
|
|
|
|
class _ProbePathFilter(logging.Filter):
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
request = getattr(record, "args", None)
|
|
if isinstance(request, dict):
|
|
path = request.get("U") or request.get("r", "")
|
|
else:
|
|
path = record.getMessage()
|
|
return not _PROBE_PATH.search(path)
|
|
|
|
|
|
_filter = _ProbePathFilter()
|
|
|
|
|
|
def on_starting(server):
|
|
logging.getLogger("gunicorn.access").addFilter(_filter)
|
|
|
|
|
|
def post_worker_init(worker):
|
|
logging.getLogger("gunicorn.access").addFilter(_filter)
|
|
from library.apps import _run_startup_probe, _should_skip_probe
|
|
|
|
if not _should_skip_probe():
|
|
try:
|
|
_run_startup_probe()
|
|
except Exception as exc:
|
|
logging.getLogger("library.apps").warning(
|
|
"Startup probe crashed: %s", exc, exc_info=True
|
|
)
|
|
|
|
|
|
def worker_exit(server, worker):
|
|
# Neomodel lazily creates a neo4j.Driver on first cypher_query and
|
|
# holds it for the process lifetime. Newer neo4j drivers warn (and
|
|
# will eventually fail to clean up) if the driver is destroyed
|
|
# without an explicit close. Close it here so each gunicorn worker
|
|
# shuts down cleanly.
|
|
try:
|
|
from neomodel import db
|
|
|
|
db.close_connection()
|
|
except Exception as exc:
|
|
logging.getLogger("neomodel").warning(
|
|
"Failed to close neomodel driver on worker exit: %s", exc
|
|
)
|