Skip to content

Commit bb3deed

Browse files
author
Carlos Sánchez
committed
fix: Defer feature-freshness thread to post-fork to avoid Gunicorn deadlock
feast_metrics.start_metrics_server() starts a feature-freshness thread in the Gunicorn master process, before Gunicorn forks its worker(s). That thread's first action, with no delay, is update_feature_freshness() -> store.list_feature_views(), which lazily builds the registry for the first time. For registry backends needing a lazy DBAPI import (e.g. the SQL registry importing pymysql via SQLAlchemy's create_engine()), this means a thread in the master can be mid-import, holding CPython's per-module import lock, at the exact moment Gunicorn forks a worker. POSIX fork() only duplicates the calling thread into the child process; every other thread in the parent, including this one, simply ceases to exist in the worker. If the fork lands while that thread holds a module's import lock, the lock stays permanently held in the new worker, since there is no longer any thread that can finish the import and release it. The worker's own later attempt to build its registry then deadlocks forever with no error - the process just hangs at "Waiting for application startup." This is intermittent by nature: it only manifests if the fork lands inside that narrow timing window. Resource monitoring already avoids this correctly (start_resource_monitoring= not uses_gunicorn plus the post_worker_init hook calling init_worker_monitoring()), but the freshness thread was not given the same treatment. This applies the identical pattern: start_metrics_server gains a start_freshness_monitoring flag (deferred exactly like resource monitoring's), and FeastServeApplication's post_worker_init hook now also calls the new init_worker_freshness_monitoring(store) after the fork, instead of feast_metrics.py starting it unconditionally beforehand. Fixes #6647 Signed-off-by: Carlos Sánchez <carlos.sancheza@cabify.com>
1 parent f9923bc commit bb3deed

3 files changed

Lines changed: 117 additions & 6 deletions

File tree

‎sdk/python/feast/feature_server.py‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1004,6 +1004,7 @@ def __init__(
10041004
store=store,
10051005
registry_ttl_sec=options["registry_ttl_sec"],
10061006
)
1007+
self._store = store
10071008
self._options = options
10081009
self._metrics_enabled = metrics_enabled
10091010
super().__init__()
@@ -1015,16 +1016,25 @@ def load_config(self):
10151016

10161017
self.cfg.set("worker_class", "uvicorn_worker.UvicornWorker")
10171018
if self._metrics_enabled:
1018-
self.cfg.set("post_worker_init", _gunicorn_post_worker_init)
1019+
store = self._store
1020+
1021+
def _post_worker_init(worker):
1022+
"""Start per-worker monitoring after Gunicorn forks.
1023+
1024+
Both resource and freshness monitoring are started here,
1025+
after the fork, rather than in the master beforehand -
1026+
see init_worker_freshness_monitoring for why starting
1027+
the freshness thread pre-fork can deadlock a worker.
1028+
"""
1029+
feast_metrics.init_worker_monitoring()
1030+
feast_metrics.init_worker_freshness_monitoring(store)
1031+
1032+
self.cfg.set("post_worker_init", _post_worker_init)
10191033
self.cfg.set("child_exit", _gunicorn_child_exit)
10201034

10211035
def load(self):
10221036
return self._app
10231037

1024-
def _gunicorn_post_worker_init(worker):
1025-
"""Start per-worker resource monitoring after Gunicorn forks."""
1026-
feast_metrics.init_worker_monitoring()
1027-
10281038
def _gunicorn_child_exit(server, worker):
10291039
"""Clean up Prometheus metric files for a dead worker."""
10301040
feast_metrics.mark_process_dead(worker.pid)
@@ -1061,6 +1071,7 @@ def start_server(
10611071
store,
10621072
metrics_config=flags,
10631073
start_resource_monitoring=not uses_gunicorn,
1074+
start_freshness_monitoring=not uses_gunicorn,
10641075
)
10651076

10661077
logger.debug("start_server called")

‎sdk/python/feast/metrics.py‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -572,11 +572,33 @@ def init_worker_monitoring():
572572
t.start()
573573

574574

575+
def init_worker_freshness_monitoring(store: "FeatureStore"):
576+
"""Start feature-freshness monitoring inside a Gunicorn worker process.
577+
578+
Called from the ``post_worker_init`` hook, after Gunicorn has forked,
579+
for the same reason ``init_worker_monitoring`` is: starting this thread
580+
in the master *before* the fork can deadlock a worker forever. The
581+
freshness thread's first action builds the registry (e.g. via
582+
``store.list_feature_views``), which for backends like the SQL registry
583+
lazily imports a DBAPI driver through SQLAlchemy's ``create_engine()``.
584+
If Gunicorn forks while that import is in progress, the driver's
585+
module import lock stays permanently held in the child, since POSIX
586+
``fork()`` only carries the calling thread into the new process - the
587+
thread that held the lock simply doesn't exist there anymore, so
588+
nothing can ever release it, and the worker's own later attempt to
589+
build its registry hangs forever with no error.
590+
"""
591+
if _config.freshness:
592+
t = threading.Thread(target=monitor_freshness, args=(store, 30), daemon=True)
593+
t.start()
594+
595+
575596
def start_metrics_server(
576597
store: "FeatureStore",
577598
port: int = 8000,
578599
metrics_config: Optional["_MetricsFlags"] = None,
579600
start_resource_monitoring: bool = True,
601+
start_freshness_monitoring: bool = True,
580602
):
581603
"""
582604
Start the Prometheus metrics HTTP server and background monitoring threads.
@@ -593,6 +615,11 @@ def start_metrics_server(
593615
monitoring thread. Set to ``False`` when Gunicorn will
594616
fork workers — the ``post_worker_init`` hook starts
595617
per-worker monitoring instead.
618+
start_freshness_monitoring: Whether to start the feature-freshness
619+
thread here. Set to ``False`` when Gunicorn will fork
620+
workers — the ``post_worker_init`` hook starts per-worker
621+
freshness monitoring instead, avoiding a fork()/import-lock
622+
deadlock (see ``init_worker_freshness_monitoring``).
596623
"""
597624
global _config
598625

@@ -632,7 +659,7 @@ def start_metrics_server(
632659
)
633660
resource_thread.start()
634661

635-
if _config.freshness:
662+
if _config.freshness and start_freshness_monitoring:
636663
freshness_thread = threading.Thread(
637664
target=monitor_freshness, args=(store, 30), daemon=True
638665
)

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

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -344,6 +344,60 @@ def test_only_resource_enabled(self):
344344
)
345345

346346

347+
class TestInitWorkerFreshnessMonitoring:
348+
"""init_worker_freshness_monitoring must mirror init_worker_monitoring's
349+
gating pattern: only start the thread when its metrics category is
350+
enabled, and always as a daemon thread targeting monitor_freshness.
351+
352+
This thread must only ever be started post-fork (from the
353+
post_worker_init hook) - starting it in the Gunicorn master before the
354+
fork can deadlock a worker forever if the fork lands while this thread
355+
is mid-import building the registry for the first time. See
356+
feast.metrics.init_worker_freshness_monitoring's docstring.
357+
"""
358+
359+
def test_starts_daemon_thread_when_freshness_enabled(self):
360+
import feast.metrics as m
361+
362+
m._config = m._MetricsFlags(
363+
enabled=True,
364+
resource=False,
365+
request=False,
366+
online_features=False,
367+
push=False,
368+
materialization=False,
369+
freshness=True,
370+
)
371+
mock_store = MagicMock()
372+
373+
with patch("feast.metrics.threading.Thread") as mock_thread_cls:
374+
m.init_worker_freshness_monitoring(mock_store)
375+
376+
mock_thread_cls.assert_called_once_with(
377+
target=m.monitor_freshness, args=(mock_store, 30), daemon=True
378+
)
379+
mock_thread_cls.return_value.start.assert_called_once()
380+
381+
def test_does_not_start_thread_when_freshness_disabled(self):
382+
import feast.metrics as m
383+
384+
m._config = m._MetricsFlags(
385+
enabled=True,
386+
resource=False,
387+
request=False,
388+
online_features=False,
389+
push=False,
390+
materialization=False,
391+
freshness=False,
392+
)
393+
mock_store = MagicMock()
394+
395+
with patch("feast.metrics.threading.Thread") as mock_thread_cls:
396+
m.init_worker_freshness_monitoring(mock_store)
397+
398+
mock_thread_cls.assert_not_called()
399+
400+
347401
class TestMetricsYamlConfig:
348402
"""Verify metrics config in feature_store.yaml is respected.
349403
@@ -438,6 +492,25 @@ def test_metrics_not_started_when_config_is_none(self):
438492
mock_fm = self._call_start_server(mock_store, cli_metrics=False)
439493
mock_fm.start_metrics_server.assert_not_called()
440494

495+
def test_freshness_monitoring_deferred_same_as_resource_monitoring(self):
496+
"""start_freshness_monitoring must mirror start_resource_monitoring
497+
exactly. Both need to be deferred to post-fork on platforms that use
498+
Gunicorn, to avoid the fork()/import-lock deadlock described in
499+
init_worker_freshness_monitoring - if this ever drifts out of sync,
500+
the freshness thread starting pre-fork can hang a worker forever."""
501+
from types import SimpleNamespace
502+
503+
mock_store = MagicMock()
504+
mock_store.config = SimpleNamespace(
505+
feature_server=SimpleNamespace(metrics=SimpleNamespace(enabled=True)),
506+
)
507+
508+
mock_fm = self._call_start_server(mock_store, cli_metrics=False)
509+
_, kwargs = mock_fm.start_metrics_server.call_args
510+
assert (
511+
kwargs["start_resource_monitoring"] == kwargs["start_freshness_monitoring"]
512+
)
513+
441514

442515
class TestTrackOnlineFeaturesEntities:
443516
def test_increments_request_count(self):

0 commit comments

Comments
 (0)