feast-dev · feast-ci-bot · Jan 18, 2022 · Jan 18, 2022 · Jan 18, 2022 · Jan 18, 2022
@@ -1,3 +1,5 @@
+import traceback
+
 import click
 import uvicorn
 from fastapi import FastAPI, HTTPException, Request
@@ -59,7 +61,7 @@ def get_online_features(body=Depends(get_body)):
             )
         except Exception as e:
             # Print the original exception on the server side
-            logger.exception(e)
+            logger.exception(traceback.format_exc())
             # Raise HTTPException to return the error message to the client
             raise HTTPException(status_code=500, detail=str(e))
 

@@ -13,14 +13,17 @@
 # limitations under the License.
 import logging
 import multiprocessing
+import time
 from datetime import datetime, timedelta
+from multiprocessing import Process
 from sys import platform
 from typing import List
 
 import pandas as pd
 import pytest
 from _pytest.nodes import Item
 
+from feast import FeatureStore
 from tests.data.data_creator import create_dataset
 from tests.integration.feature_repos.integration_test_repo_config import (
     IntegrationTestRepoConfig,
@@ -137,23 +140,41 @@ def simple_dataset_2() -> pd.DataFrame:
     return pd.DataFrame.from_dict(data)
 
 
+def start_test_local_server(repo_path: str, port: int):
+    fs = FeatureStore(repo_path)
+    fs.serve("localhost", port, no_access_log=True)
+
+
 @pytest.fixture(
     params=FULL_REPO_CONFIGS, scope="session", ids=[str(c) for c in FULL_REPO_CONFIGS]
 )
-def environment(request):
-    e = construct_test_environment(request.param)
+def environment(request, worker_id: str):
+    e = construct_test_environment(request.param, worker_id=worker_id)
+    proc = Process(
+        target=start_test_local_server,
+        args=(e.feature_store.repo_path, e.get_local_server_port()),
+        daemon=True,
+    )
+    if e.python_feature_server and e.test_repo_config.provider == "local":
+        proc.start()
+        # Wait for server to start
+        time.sleep(3)
 
     def cleanup():
         e.feature_store.teardown()
+        if proc.is_alive():
+            proc.kill()
 
     request.addfinalizer(cleanup)
+
     return e
 
 
 @pytest.fixture()
 def local_redis_environment(request, worker_id):
-
-    e = construct_test_environment(IntegrationTestRepoConfig(online_store=REDIS_CONFIG))
+    e = construct_test_environment(
+        IntegrationTestRepoConfig(online_store=REDIS_CONFIG), worker_id=worker_id
+    )
 
     def cleanup():
         e.feature_store.teardown()

@@ -10,7 +10,7 @@
     name="driver_hourly_stats",  # Intentionally use the same FeatureView name
     entities=["driver_id"],
     online=False,
-    input=driver_hourly_stats,
+    batch_source=driver_hourly_stats,
     ttl=Duration(seconds=10),
     tags={},
 )
@@ -19,7 +19,7 @@
     name="driver_hourly_stats",  # Intentionally use the same FeatureView name
     entities=["driver_id"],
     online=False,
-    input=driver_hourly_stats,
+    batch_source=driver_hourly_stats,
     ttl=Duration(seconds=10),
     tags={},
 )
@@ -1,6 +1,7 @@
 import importlib
 import json
 import os
+import re
 import tempfile
 import uuid
 from dataclasses import dataclass, field
@@ -51,6 +52,7 @@
 DEFAULT_FULL_REPO_CONFIGS: List[IntegrationTestRepoConfig] = [
     # Local configurations
     IntegrationTestRepoConfig(),
+    IntegrationTestRepoConfig(python_feature_server=True),
 ]
 if os.getenv("FEAST_IS_LOCAL_TEST", "False") != "True":
     DEFAULT_FULL_REPO_CONFIGS.extend(
@@ -217,6 +219,7 @@ class Environment:
     feature_store: FeatureStore
     data_source_creator: DataSourceCreator
     python_feature_server: bool
+    worker_id: str
 
     end_date: datetime = field(
         default=datetime.utcnow().replace(microsecond=0, second=0, minute=0)
@@ -225,6 +228,20 @@ class Environment:
     def __post_init__(self):
         self.start_date: datetime = self.end_date - timedelta(days=3)
 
+    def get_feature_server_endpoint(self) -> str:
+        if self.python_feature_server and self.test_repo_config.provider == "local":
+            return f"http://localhost:{self.get_local_server_port()}"
+        return self.feature_store.get_feature_server_endpoint()
+
+    def get_local_server_port(self) -> int:
+        # Heuristic when running with xdist to extract unique ports for each worker
+        parsed_worker_id = re.findall("gw(\\d+)", self.worker_id)
+        if len(parsed_worker_id) != 0:
+            worker_id_num = int(parsed_worker_id[0])
+        else:
+            worker_id_num = 0
+        return 6566 + worker_id_num
+
 
 def table_name_from_data_source(ds: DataSource) -> Optional[str]:
     if hasattr(ds, "table_ref"):
@@ -237,6 +254,7 @@ def table_name_from_data_source(ds: DataSource) -> Optional[str]:
 def construct_test_environment(
     test_repo_config: IntegrationTestRepoConfig,
     test_suite_name: str = "integration_test",
+    worker_id: str = "worker_id",
 ) -> Environment:
 
     _uuid = str(uuid.uuid4()).replace("-", "")[:8]
@@ -254,7 +272,7 @@ def construct_test_environment(
 
     repo_dir_name = tempfile.mkdtemp()
 
-    if test_repo_config.python_feature_server:
+    if test_repo_config.python_feature_server and test_repo_config.provider == "aws":
         from feast.infra.feature_servers.aws_lambda.config import (
             AwsLambdaFeatureServerConfig,
         )
@@ -266,6 +284,7 @@ def construct_test_environment(
 
         registry = f"s3://feast-integration-tests/registries/{project}/registry.db"
     else:
+        # Note: even if it's a local feature server, the repo config does not have this configured
         feature_server = None
         registry = str(Path(repo_dir_name) / "registry.db")
 
@@ -293,6 +312,7 @@ def construct_test_environment(
         feature_store=fs,
         data_source_creator=offline_creator,
         python_feature_server=test_repo_config.python_feature_server,
+        worker_id=worker_id,
     )
 
     return environment
@@ -20,7 +20,7 @@ def driver_feature_view(
         entities=["driver"],
         features=None if infer_features else [Feature("value", value_type)],
         ttl=timedelta(days=5),
-        input=data_source,
+        batch_source=data_source,
     )
 
 
@@ -35,7 +35,7 @@ def global_feature_view(
         entities=[],
         features=None if infer_features else [Feature("entityless_value", value_type)],
         ttl=timedelta(days=5),
-        input=data_source,
+        batch_source=data_source,
     )
 
 

@@ -185,7 +185,7 @@ def _get_online_features_dict_remotely(
 
     The output should be identical to:
 
-    >>> fs.get_online_features(features=features, entity_rows=entity_rows, full_feature_names=full_feature_names).to_dict()
+    fs.get_online_features(features=features, entity_rows=entity_rows, full_feature_names=full_feature_names).to_dict()
 
     This makes it easy to test the remote feature server by comparing the output to the local method.
 
@@ -212,6 +212,10 @@ def _get_online_features_dict_remotely(
         time.sleep(1)
     else:
         raise Exception("Failed to get online features from remote feature server")
+    if "metadata" not in response:
+        raise Exception(
+            f"Failed to get online features from remote feature server {response}"
+        )
     keys = response["metadata"]["feature_names"]
     # Get rid of unnecessary structure in the response, leaving list of dicts
     response = [row["values"] for row in response["results"]]
@@ -238,8 +242,8 @@ def get_online_features_dict(
     assertpy.assert_that(online_features).is_not_none()
     dict1 = online_features.to_dict()
 
-    endpoint = environment.feature_store.get_feature_server_endpoint()
-    # If endpoint is None, it means that the remote feature server isn't configured
+    endpoint = environment.get_feature_server_endpoint()
+    # If endpoint is None, it means that a local / remote feature server aren't configured
     if endpoint is not None:
         dict2 = _get_online_features_dict_remotely(
             endpoint=endpoint,