Change the feast serve endpoint to be sync rather than async. (feast-dev#2119)

nossrannug · web-flow · commit 7d4369f0b1ce · 2021-12-17T12:42:54.000-08:00
When an endpoint is async, everything happens on the event loop thread.
This is great for functionality that is IO heavy since the job can be
awaited and other requests processed in the meantime. But if there are
no awaits, then everything happens synchronously on one thread which
means that while we're waiting for a response from a DB or API call,
nothing else gets processed.
To address this I've removed the async keyword from the endpoint.
This makes FastAPI process each request on a thread in a thread pool.

Signed-off-by: Gunnar Sv Sigurbjörnsson &lt;gunnar.sigurbjornsson@gmail.com&gt;
diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py
@@ -508,14 +508,17 @@ def init_command(project_directory, minimal: bool, template: str):
     default=6566,
     help="Specify a port for the server [default: 6566]",
 )
+@click.option(
+    "--no-access-log", is_flag=True, help="Disable the Uvicorn access log.",
+)
 @click.pass_context
-def serve_command(ctx: click.Context, host: str, port: int):
+def serve_command(ctx: click.Context, host: str, port: int, no_access_log: bool):
     """[Experimental] Start a the feature consumption server locally on a given port."""
     repo = ctx.obj["CHDIR"]
     cli_check_repo(repo)
     store = FeatureStore(repo_path=str(repo))
 
-    store.serve(host, port)
+    store.serve(host, port, no_access_log)
 
 
 @cli.command("serve_transformations")
diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py
@@ -2,6 +2,7 @@
 import uvicorn
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.logger import logger
+from fastapi.params import Depends
 from google.protobuf.json_format import MessageToDict, Parse
 
 import feast
@@ -15,17 +16,21 @@ def get_app(store: "feast.FeatureStore"):
 
     app = FastAPI()
 
+    async def get_body(request: Request):
+        return await request.body()
+
     @app.post("/get-online-features")
-    async def get_online_features(request: Request):
+    def get_online_features(body=Depends(get_body)):
         try:
             # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object
-            body = await request.body()
             request_proto = GetOnlineFeaturesRequest()
             Parse(body, request_proto)
 
             # Initialize parameters for FeatureStore.get_online_features(...) call
             if request_proto.HasField("feature_service"):
-                features = store.get_feature_service(request_proto.feature_service)
+                features = store.get_feature_service(
+                    request_proto.feature_service, allow_cache=True
+                )
             else:
                 features = list(request_proto.features.val)
 
@@ -61,11 +66,13 @@ async def get_online_features(request: Request):
     return app
 
 
-def start_server(store: "feast.FeatureStore", host: str, port: int):
+def start_server(
+    store: "feast.FeatureStore", host: str, port: int, no_access_log: bool
+):
     app = get_app(store)
     click.echo(
         "This is an "
         + click.style("experimental", fg="yellow", bold=True, underline=True)
         + " feature. It's intended for early testing and feedback, and could change without warnings in future releases."
     )
-    uvicorn.run(app, host=host, port=port)
+    uvicorn.run(app, host=host, port=port, access_log=(not no_access_log))
diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py
@@ -288,7 +288,9 @@ def get_entity(self, name: str) -> Entity:
         return self._registry.get_entity(name, self.project)
 
     @log_exceptions_and_usage
-    def get_feature_service(self, name: str) -> FeatureService:
+    def get_feature_service(
+        self, name: str, allow_cache: bool = False
+    ) -> FeatureService:
         """
         Retrieves a feature service.
 
@@ -301,7 +303,7 @@ def get_feature_service(self, name: str) -> FeatureService:
         Raises:
             FeatureServiceNotFoundException: The feature service could not be found.
         """
-        return self._registry.get_feature_service(name, self.project)
+        return self._registry.get_feature_service(name, self.project, allow_cache)
 
     @log_exceptions_and_usage
     def get_feature_view(self, name: str) -> FeatureView:
@@ -369,14 +371,19 @@ def delete_feature_service(self, name: str):
             """
         return self._registry.delete_feature_service(name, self.project)
 
-    def _get_features(self, features: Union[List[str], FeatureService],) -> List[str]:
+    def _get_features(
+        self, features: Union[List[str], FeatureService], allow_cache: bool = False,
+    ) -> List[str]:
         _features = features
+
         if not _features:
             raise ValueError("No features specified for retrieval")
 
         _feature_refs = []
         if isinstance(_features, FeatureService):
-            feature_service_from_registry = self.get_feature_service(_features.name)
+            feature_service_from_registry = self.get_feature_service(
+                _features.name, allow_cache
+            )
             if feature_service_from_registry != _features:
                 warnings.warn(
                     "The FeatureService object that has been passed in as an argument is"
@@ -1040,7 +1047,7 @@ def get_online_features(
             ... )
             >>> online_response_dict = online_response.to_dict()
         """
-        _feature_refs = self._get_features(features)
+        _feature_refs = self._get_features(features, allow_cache=True)
         (
             requested_feature_views,
             requested_request_feature_views,
@@ -1455,12 +1462,12 @@ def _get_feature_views_to_use(
         return views_to_use
 
     @log_exceptions_and_usage
-    def serve(self, host: str, port: int) -> None:
+    def serve(self, host: str, port: int, no_access_log: bool) -> None:
         """Start the feature consumption server locally on a given port."""
         if not flags_helper.enable_python_feature_server(self.config):
             raise ExperimentalFeatureNotEnabled(flags.FLAG_PYTHON_FEATURE_SERVER_NAME)
 
-        feature_server.start_server(self, host, port)
+        feature_server.start_server(self, host, port, no_access_log)
 
     @log_exceptions_and_usage
     def get_feature_server_endpoint(self) -> Optional[str]:
diff --git a/sdk/python/feast/registry.py b/sdk/python/feast/registry.py
@@ -15,6 +15,7 @@
 from collections import defaultdict
 from datetime import datetime, timedelta
 from pathlib import Path
+from threading import Lock
 from typing import Any, Dict, List, Optional
 from urllib.parse import urlparse
 
@@ -101,7 +102,6 @@ class Registry:
     cached_registry_proto: Optional[RegistryProto] = None
     cached_registry_proto_created: Optional[datetime] = None
     cached_registry_proto_ttl: timedelta
-    cache_being_updated: bool = False
 
     def __init__(
         self, registry_config: Optional[RegistryConfig], repo_path: Optional[Path]
@@ -115,6 +115,8 @@ def __init__(
             or where it will be created if it does not exist yet.
         """
 
+        self._refresh_lock = Lock()
+
         if registry_config:
             registry_store_type = registry_config.registry_store_type
             registry_path = registry_config.path
@@ -325,6 +327,7 @@ def get_feature_service(
         Args:
             name: Name of feature service
             project: Feast project that this feature service belongs to
+            allow_cache: Whether to allow returning this feature service from a cached registry
 
         Returns:
             Returns either the specified feature service, or raises an exception if
@@ -347,6 +350,7 @@ def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Enti
         Args:
             name: Name of entity
             project: Feast project that this entity belongs to
+            allow_cache: Whether to allow returning this entity from a cached registry
 
         Returns:
             Returns either the specified entity, or raises an exception if
@@ -419,8 +423,8 @@ def list_on_demand_feature_views(
         Retrieve a list of on demand feature views from the registry
 
         Args:
-            allow_cache: Whether to allow returning on demand feature views from a cached registry
             project: Filter on demand feature views based on project name
+            allow_cache: Whether to allow returning on demand feature views from a cached registry
 
         Returns:
             List of on demand feature views
@@ -733,31 +737,31 @@ def _get_registry_proto(self, allow_cache: bool = False) -> RegistryProto:
 
         Returns: Returns a RegistryProto object which represents the state of the registry
         """
-        expired = (
-            self.cached_registry_proto is None
-            or self.cached_registry_proto_created is None
-        ) or (
-            self.cached_registry_proto_ttl.total_seconds() > 0  # 0 ttl means infinity
-            and (
-                datetime.now()
-                > (self.cached_registry_proto_created + self.cached_registry_proto_ttl)
+        with self._refresh_lock:
+            expired = (
+                self.cached_registry_proto is None
+                or self.cached_registry_proto_created is None
+            ) or (
+                self.cached_registry_proto_ttl.total_seconds()
+                > 0  # 0 ttl means infinity
+                and (
+                    datetime.now()
+                    > (
+                        self.cached_registry_proto_created
+                        + self.cached_registry_proto_ttl
+                    )
+                )
             )
-        )
 
-        if allow_cache and (not expired or self.cache_being_updated):
-            assert isinstance(self.cached_registry_proto, RegistryProto)
-            return self.cached_registry_proto
+            if allow_cache and not expired:
+                assert isinstance(self.cached_registry_proto, RegistryProto)
+                return self.cached_registry_proto
 
-        try:
-            self.cache_being_updated = True
             registry_proto = self._registry_store.get_registry_proto()
             self.cached_registry_proto = registry_proto
             self.cached_registry_proto_created = datetime.now()
-        except Exception as e:
-            raise e
-        finally:
-            self.cache_being_updated = False
-        return registry_proto
+
+            return registry_proto
 
     def _check_conflicting_feature_view_names(self, feature_view: BaseFeatureView):
         name_to_fv_protos = self._existing_feature_view_names_to_fvs()
