Skip to content

Commit 68c5d54

Browse files
aborgatinclaude
andauthored
fix: Keep the metrics server running when a Gunicorn worker exits (#7012)
* fix: Keep the metrics server running when a Gunicorn worker exits Since the metrics server runs in a multiprocessing.Process started in the Gunicorn master, every forked worker inherits multiprocessing's registry of child processes. At worker exit, multiprocessing.util._exit_function terminates every daemon process in that registry, so the first worker restart (--max-requests, timeout, reload) sends SIGTERM to the master's metrics server and /metrics stops responding. It also logs "AssertionError: can only join a child process" on every worker exit. Register an at-fork hook when the metrics process is started that removes it from the registry in every forked child. Workers no longer terminate it; the master still does on its own exit. Fixes #7011 Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Alexandr Borgatin <[email protected]> * test: Wait for the metrics server before forking in the worker-exit test On macOS CI the spawned metrics process can take longer than the 5 s readiness wait to start listening, so the first /metrics request was refused. Poll /metrics (up to 30 s) before forking the child. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Alexandr Borgatin <[email protected]> * test: Use an instant stand-in process in the worker-exit test The spawned metrics server did not come up within 30 s inside the full unit suite on macOS CI. The test is about the process surviving a forked child's exit, not about serving HTTP, so register a fork-context daemon process via start_metrics_server instead and assert it is still alive after the child runs multiprocessing.util._exit_function(). Without the fix it is terminated (exitcode -SIGTERM). Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Alexandr Borgatin <[email protected]> * test: Check active_children() in the forked child instead of running _exit_function Running multiprocessing.util._exit_function() in a fork of the pytest process runs every registered finalizer. Check what a worker's exit would act on instead: the metrics process must not be in the child's active_children(). Also cover _forget_metrics_process_in_child directly, and kill the stand-in process in cleanup (terminate() can wait out the join timeout under xdist). Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Alexandr Borgatin <[email protected]> --------- Signed-off-by: Alexandr Borgatin <[email protected]> Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
1 parent 1299e4e commit 68c5d54

2 files changed

Lines changed: 89 additions & 1 deletion

File tree

‎sdk/python/feast/metrics.py‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -681,6 +681,27 @@ def _shutdown(signum, frame):
681681
httpd.serve_forever()
682682

683683

684+
# The metrics-server process started by start_metrics_server() in this process
685+
# (the Gunicorn master), and whether the at-fork hook below is registered.
686+
_metrics_process: Optional[multiprocessing.Process] = None
687+
_metrics_at_fork_registered: bool = False
688+
689+
690+
def _forget_metrics_process_in_child() -> None:
691+
"""Drop the inherited metrics-server process from multiprocessing's registry.
692+
693+
Runs in every child forked from the process that started the metrics
694+
server (for example each Gunicorn worker). A forked child inherits
695+
multiprocessing's registry of child processes, and at interpreter exit
696+
``multiprocessing.util._exit_function`` terminates every daemon process in
697+
it. Without this, the first worker exit would SIGTERM the master's metrics
698+
server and ``/metrics`` would stop responding. The child does not own the
699+
process; the parent still terminates it on its own exit.
700+
"""
701+
if _metrics_process is not None:
702+
multiprocessing.process._children.discard(_metrics_process) # type: ignore[attr-defined]
703+
704+
684705
def start_metrics_server(
685706
store: "FeatureStore",
686707
port: int = 8000,
@@ -714,7 +735,7 @@ def start_metrics_server(
714735
workers — the ``post_worker_init`` hook starts per-worker
715736
freshness monitoring instead.
716737
"""
717-
global _config
738+
global _config, _metrics_process, _metrics_at_fork_registered
718739

719740
if metrics_config is not None:
720741
_config = metrics_config
@@ -739,6 +760,10 @@ def start_metrics_server(
739760
name="feast-metrics-server",
740761
)
741762
metrics_proc.start()
763+
_metrics_process = metrics_proc
764+
if not _metrics_at_fork_registered and hasattr(os, "register_at_fork"):
765+
os.register_at_fork(after_in_child=_forget_metrics_process_in_child)
766+
_metrics_at_fork_registered = True
742767

743768
if ready_event.wait(timeout=5):
744769
logger.info(

‎sdk/python/tests/unit/test_metrics.py‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,10 @@
1313
# limitations under the License.
1414

1515
import http.client
16+
import os
1617
import socket
1718
import threading
19+
import time
1820
from datetime import datetime, timedelta, timezone
1921
from unittest.mock import MagicMock, patch
2022

@@ -642,6 +644,67 @@ def test_ready_event_set_on_successful_bind(self):
642644
mock_event.set.assert_called_once()
643645

644646

647+
@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
648+
@pytest.mark.filterwarnings(
649+
"ignore:This process .* is multi-threaded:DeprecationWarning"
650+
)
651+
class TestMetricsProcessSurvivesForkedChildExit:
652+
"""A forked child (e.g. a Gunicorn worker) must not terminate the metrics server on exit."""
653+
654+
def test_forked_child_does_not_track_metrics_process(self):
655+
import multiprocessing
656+
657+
import feast.metrics as m
658+
659+
# Stand-in for the metrics-server process: a real daemon process that
660+
# starts instantly on every platform, registered by start_metrics_server.
661+
proc = multiprocessing.get_context("fork").Process(
662+
target=time.sleep, args=(60,), daemon=True
663+
)
664+
mock_event = MagicMock()
665+
mock_event.wait.return_value = True
666+
667+
with (
668+
patch("feast.metrics.multiprocessing.Process", return_value=proc),
669+
patch("feast.metrics.multiprocessing.Event", return_value=mock_event),
670+
):
671+
m.start_metrics_server(
672+
MagicMock(),
673+
start_resource_monitoring=False,
674+
start_freshness_monitoring=False,
675+
)
676+
try:
677+
pid = os.fork()
678+
if pid == 0:
679+
# A worker's interpreter exit terminates every daemon process in
680+
# active_children(); report whether the metrics process is in it.
681+
tracked = proc in multiprocessing.active_children()
682+
os._exit(1 if tracked else 0)
683+
_, status = os.waitpid(pid, 0)
684+
685+
assert os.waitstatus_to_exitcode(status) == 0
686+
assert proc.is_alive()
687+
finally:
688+
proc.kill()
689+
proc.join(timeout=5)
690+
m._metrics_process = None
691+
692+
693+
def test_forget_metrics_process_in_child_removes_it_from_registry():
694+
import multiprocessing
695+
696+
import feast.metrics as m
697+
698+
proc = MagicMock()
699+
children = {proc}
700+
with (
701+
patch.object(multiprocessing.process, "_children", children),
702+
patch.object(m, "_metrics_process", proc),
703+
):
704+
m._forget_metrics_process_in_child()
705+
assert proc not in children
706+
707+
645708
class TestMetricsYamlConfig:
646709
"""Verify metrics config in feature_store.yaml is respected.
647710

0 commit comments

Comments
 (0)