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):