From 537675154ff8351f05c1c5dfff79adff7a6ede73 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20S=C3=A1nchez?= Date: Mon, 27 Jul 2026 15:26:07 +0200 Subject: [PATCH] fix: Defer feature-freshness thread to post-fork to avoid Gunicorn deadlock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- sdk/python/feast/feature_server.py | 13 ++++- sdk/python/feast/metrics.py | 18 +++++- sdk/python/tests/unit/test_feature_server.py | 17 +++++- sdk/python/tests/unit/test_metrics.py | 60 ++++++++++++++++++++ 4 files changed, 103 insertions(+), 5 deletions(-) diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index 1f374236790..7412931ce8a 100644 --- a/sdk/python/feast/feature_server.py +++ b/sdk/python/feast/feature_server.py @@ -13,6 +13,7 @@ # limitations under the License. import asyncio +import functools import os import sys import threading @@ -1004,6 +1005,7 @@ def __init__( store=store, registry_ttl_sec=options["registry_ttl_sec"], ) + self._store = store self._options = options self._metrics_enabled = metrics_enabled super().__init__() @@ -1015,15 +1017,19 @@ def load_config(self): self.cfg.set("worker_class", "uvicorn_worker.UvicornWorker") if self._metrics_enabled: - self.cfg.set("post_worker_init", _gunicorn_post_worker_init) + self.cfg.set( + "post_worker_init", + functools.partial(_gunicorn_post_worker_init, self._store), + ) self.cfg.set("child_exit", _gunicorn_child_exit) def load(self): return self._app - def _gunicorn_post_worker_init(worker): - """Start per-worker resource monitoring after Gunicorn forks.""" + def _gunicorn_post_worker_init(store: "feast.FeatureStore", worker): + """Start per-worker resource and freshness monitoring after Gunicorn forks.""" feast_metrics.init_worker_monitoring() + feast_metrics.init_worker_freshness_monitoring(store) def _gunicorn_child_exit(server, worker): """Clean up Prometheus metric files for a dead worker.""" @@ -1061,6 +1067,7 @@ def start_server( store, metrics_config=flags, start_resource_monitoring=not uses_gunicorn, + start_freshness_monitoring=not uses_gunicorn, ) logger.debug("start_server called") diff --git a/sdk/python/feast/metrics.py b/sdk/python/feast/metrics.py index 13a855d587b..85c0ceadabe 100644 --- a/sdk/python/feast/metrics.py +++ b/sdk/python/feast/metrics.py @@ -572,11 +572,23 @@ def init_worker_monitoring(): t.start() +def init_worker_freshness_monitoring(store: "FeatureStore"): + """Start feature-freshness monitoring inside a Gunicorn worker process. + + Called from the ``post_worker_init`` hook so that each worker starts + its own freshness monitoring after the fork, not before. + """ + if _config.freshness: + t = threading.Thread(target=monitor_freshness, args=(store, 30), daemon=True) + t.start() + + def start_metrics_server( store: "FeatureStore", port: int = 8000, metrics_config: Optional["_MetricsFlags"] = None, start_resource_monitoring: bool = True, + start_freshness_monitoring: bool = True, ): """ Start the Prometheus metrics HTTP server and background monitoring threads. @@ -593,6 +605,10 @@ def start_metrics_server( monitoring thread. Set to ``False`` when Gunicorn will fork workers — the ``post_worker_init`` hook starts per-worker monitoring instead. + start_freshness_monitoring: Whether to start the feature-freshness + thread here. Set to ``False`` when Gunicorn will fork + workers — the ``post_worker_init`` hook starts per-worker + freshness monitoring instead. """ global _config @@ -632,7 +648,7 @@ def start_metrics_server( ) resource_thread.start() - if _config.freshness: + if _config.freshness and start_freshness_monitoring: freshness_thread = threading.Thread( target=monitor_freshness, args=(store, 30), daemon=True ) diff --git a/sdk/python/tests/unit/test_feature_server.py b/sdk/python/tests/unit/test_feature_server.py index 0fef1aea463..b9f2afd4a61 100644 --- a/sdk/python/tests/unit/test_feature_server.py +++ b/sdk/python/tests/unit/test_feature_server.py @@ -12,10 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. import json +import sys import time from collections import Counter from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi.testclient import TestClient @@ -749,3 +750,17 @@ def test_metadata_model_accepts_raw_proto_dict(): ) assert full.metadata is not None assert full.metadata.feature_names == ["x", "y"] + + +@pytest.mark.skipif(sys.platform == "win32", reason="Gunicorn is not used on Windows") +def test_gunicorn_post_worker_init_starts_resource_and_freshness_monitoring(): + from feast.feature_server import _gunicorn_post_worker_init + + mock_store = MagicMock() + with ( + patch("feast.feature_server.feast_metrics") as mock_fm, + ): + _gunicorn_post_worker_init(mock_store, worker=MagicMock()) + + mock_fm.init_worker_monitoring.assert_called_once() + mock_fm.init_worker_freshness_monitoring.assert_called_once_with(mock_store) diff --git a/sdk/python/tests/unit/test_metrics.py b/sdk/python/tests/unit/test_metrics.py index 6099fabb46c..a3a65ebe782 100644 --- a/sdk/python/tests/unit/test_metrics.py +++ b/sdk/python/tests/unit/test_metrics.py @@ -344,6 +344,51 @@ def test_only_resource_enabled(self): ) +class TestInitWorkerFreshnessMonitoring: + """init_worker_freshness_monitoring mirrors init_worker_monitoring's gating.""" + + def test_starts_daemon_thread_when_freshness_enabled(self): + import feast.metrics as m + + m._config = m._MetricsFlags( + enabled=True, + resource=False, + request=False, + online_features=False, + push=False, + materialization=False, + freshness=True, + ) + mock_store = MagicMock() + + with patch("feast.metrics.threading.Thread") as mock_thread_cls: + m.init_worker_freshness_monitoring(mock_store) + + mock_thread_cls.assert_called_once_with( + target=m.monitor_freshness, args=(mock_store, 30), daemon=True + ) + mock_thread_cls.return_value.start.assert_called_once() + + def test_does_not_start_thread_when_freshness_disabled(self): + import feast.metrics as m + + m._config = m._MetricsFlags( + enabled=True, + resource=False, + request=False, + online_features=False, + push=False, + materialization=False, + freshness=False, + ) + mock_store = MagicMock() + + with patch("feast.metrics.threading.Thread") as mock_thread_cls: + m.init_worker_freshness_monitoring(mock_store) + + mock_thread_cls.assert_not_called() + + class TestMetricsYamlConfig: """Verify metrics config in feature_store.yaml is respected. @@ -438,6 +483,21 @@ def test_metrics_not_started_when_config_is_none(self): mock_fm = self._call_start_server(mock_store, cli_metrics=False) mock_fm.start_metrics_server.assert_not_called() + def test_freshness_monitoring_deferred_same_as_resource_monitoring(self): + """start_freshness_monitoring must mirror start_resource_monitoring exactly.""" + from types import SimpleNamespace + + mock_store = MagicMock() + mock_store.config = SimpleNamespace( + feature_server=SimpleNamespace(metrics=SimpleNamespace(enabled=True)), + ) + + mock_fm = self._call_start_server(mock_store, cli_metrics=False) + _, kwargs = mock_fm.start_metrics_server.call_args + assert ( + kwargs["start_resource_monitoring"] == kwargs["start_freshness_monitoring"] + ) + class TestTrackOnlineFeaturesEntities: def test_increments_request_count(self):