Skip to content

Commit 597a845

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 597a845

3 files changed

Lines changed: 87 additions & 6 deletions

File tree

‎sdk/python/feast/feature_server.py‎

Lines changed: 10 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,19 @@ 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 resource and freshness monitoring after Gunicorn forks."""
1023+
feast_metrics.init_worker_monitoring()
1024+
feast_metrics.init_worker_freshness_monitoring(store)
1025+
1026+
self.cfg.set("post_worker_init", _post_worker_init)
10191027
self.cfg.set("child_exit", _gunicorn_child_exit)
10201028

10211029
def load(self):
10221030
return self._app
10231031

1024-
def _gunicorn_post_worker_init(worker):
1025-
"""Start per-worker resource monitoring after Gunicorn forks."""
1026-
feast_metrics.init_worker_monitoring()
1027-
10281032
def _gunicorn_child_exit(server, worker):
10291033
"""Clean up Prometheus metric files for a dead worker."""
10301034
feast_metrics.mark_process_dead(worker.pid)
@@ -1061,6 +1065,7 @@ def start_server(
10611065
store,
10621066
metrics_config=flags,
10631067
start_resource_monitoring=not uses_gunicorn,
1068+
start_freshness_monitoring=not uses_gunicorn,
10641069
)
10651070

10661071
logger.debug("start_server called")

‎sdk/python/feast/metrics.py‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -572,11 +572,23 @@ 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 so that each worker starts
579+
its own freshness monitoring after the fork, not before.
580+
"""
581+
if _config.freshness:
582+
t = threading.Thread(target=monitor_freshness, args=(store, 30), daemon=True)
583+
t.start()
584+
585+
575586
def start_metrics_server(
576587
store: "FeatureStore",
577588
port: int = 8000,
578589
metrics_config: Optional["_MetricsFlags"] = None,
579590
start_resource_monitoring: bool = True,
591+
start_freshness_monitoring: bool = True,
580592
):
581593
"""
582594
Start the Prometheus metrics HTTP server and background monitoring threads.
@@ -593,6 +605,10 @@ def start_metrics_server(
593605
monitoring thread. Set to ``False`` when Gunicorn will
594606
fork workers — the ``post_worker_init`` hook starts
595607
per-worker monitoring instead.
608+
start_freshness_monitoring: Whether to start the feature-freshness
609+
thread here. Set to ``False`` when Gunicorn will fork
610+
workers — the ``post_worker_init`` hook starts per-worker
611+
freshness monitoring instead.
596612
"""
597613
global _config
598614

@@ -632,7 +648,7 @@ def start_metrics_server(
632648
)
633649
resource_thread.start()
634650

635-
if _config.freshness:
651+
if _config.freshness and start_freshness_monitoring:
636652
freshness_thread = threading.Thread(
637653
target=monitor_freshness, args=(store, 30), daemon=True
638654
)

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

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

346346

347+
class TestInitWorkerFreshnessMonitoring:
348+
"""init_worker_freshness_monitoring mirrors init_worker_monitoring's gating."""
349+
350+
def test_starts_daemon_thread_when_freshness_enabled(self):
351+
import feast.metrics as m
352+
353+
m._config = m._MetricsFlags(
354+
enabled=True,
355+
resource=False,
356+
request=False,
357+
online_features=False,
358+
push=False,
359+
materialization=False,
360+
freshness=True,
361+
)
362+
mock_store = MagicMock()
363+
364+
with patch("feast.metrics.threading.Thread") as mock_thread_cls:
365+
m.init_worker_freshness_monitoring(mock_store)
366+
367+
mock_thread_cls.assert_called_once_with(
368+
target=m.monitor_freshness, args=(mock_store, 30), daemon=True
369+
)
370+
mock_thread_cls.return_value.start.assert_called_once()
371+
372+
def test_does_not_start_thread_when_freshness_disabled(self):
373+
import feast.metrics as m
374+
375+
m._config = m._MetricsFlags(
376+
enabled=True,
377+
resource=False,
378+
request=False,
379+
online_features=False,
380+
push=False,
381+
materialization=False,
382+
freshness=False,
383+
)
384+
mock_store = MagicMock()
385+
386+
with patch("feast.metrics.threading.Thread") as mock_thread_cls:
387+
m.init_worker_freshness_monitoring(mock_store)
388+
389+
mock_thread_cls.assert_not_called()
390+
391+
347392
class TestMetricsYamlConfig:
348393
"""Verify metrics config in feature_store.yaml is respected.
349394
@@ -438,6 +483,21 @@ def test_metrics_not_started_when_config_is_none(self):
438483
mock_fm = self._call_start_server(mock_store, cli_metrics=False)
439484
mock_fm.start_metrics_server.assert_not_called()
440485

486+
def test_freshness_monitoring_deferred_same_as_resource_monitoring(self):
487+
"""start_freshness_monitoring must mirror start_resource_monitoring exactly."""
488+
from types import SimpleNamespace
489+
490+
mock_store = MagicMock()
491+
mock_store.config = SimpleNamespace(
492+
feature_server=SimpleNamespace(metrics=SimpleNamespace(enabled=True)),
493+
)
494+
495+
mock_fm = self._call_start_server(mock_store, cli_metrics=False)
496+
_, kwargs = mock_fm.start_metrics_server.call_args
497+
assert (
498+
kwargs["start_resource_monitoring"] == kwargs["start_freshness_monitoring"]
499+
)
500+
441501

442502
class TestTrackOnlineFeaturesEntities:
443503
def test_increments_request_count(self):

0 commit comments

Comments
 (0)