Feature transformation server docker image (#1972)

* Add Dockerfile and app for building FTS docker image

Signed-off-by: Felix Wang <wangfelix98@gmail.com>

* Add CI for building FTS

Signed-off-by: Felix Wang <wangfelix98@gmail.com>

.github/workflows/build_feature_transformation_server.yml

@@ -0,0 +1,28 @@
+name: build_feature_transformation_server
+
+on:
+  push:
+    branches:
+      - master
+
+jobs:
+  build-docker-image:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v1
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v1
+      - name: Login to DockerHub
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Build and push
+        run: |
+          docker build \
+            --file sdk/python/feast/infra/transformation_servers/Dockerfile \
+            --tag feastdev/feature-transformation-server:`git rev-parse HEAD` \
+            .
+          docker push feastdev/feature-transformation-server:`git rev-parse HEAD`

sdk/python/feast/cli.py

@@ -23,6 +23,7 @@
 from colorama import Fore, Style
 
 from feast import flags, flags_helper, utils
+from feast.constants import DEFAULT_FEATURE_TRANSFORMATION_SERVER_PORT
 from feast.errors import FeastObjectNotFoundException, FeastProviderLoginError
 from feast.feature_store import FeatureStore
 from feast.feature_view import FeatureView
@@ -477,7 +478,11 @@ def serve_command(ctx: click.Context, port: int):
 
 @cli.command("serve_transformations")
 @click.option(
-    "--port", "-p", type=click.INT, default=6569, help="Specify a port for the server"
+    "--port",
+    "-p",
+    type=click.INT,
+    default=DEFAULT_FEATURE_TRANSFORMATION_SERVER_PORT,
+    help="Specify a port for the server",
 )
 @click.pass_context
 def serve_transformations_command(ctx: click.Context, port: int):

sdk/python/feast/constants.py

@@ -22,8 +22,17 @@
 # feature_store.yaml environment variable name for remote feature server
 FEATURE_STORE_YAML_ENV_NAME: str = "FEATURE_STORE_YAML_BASE64"
 
+# Environment variable for registry
+REGISTRY_ENV_NAME: str = "REGISTRY_BASE64"
+
 # Environment variable for toggling usage
 FEAST_USAGE = "FEAST_USAGE"
 
 # Environment variable for the path for overwriting universal test configs
 FULL_REPO_CONFIGS_MODULE_ENV_NAME: str = "FULL_REPO_CONFIGS_MODULE"
+
+# Environment variable for overwriting FTS port
+FEATURE_TRANSFORMATION_SERVER_PORT_ENV_NAME: str = "FEATURE_TRANSFORMATION_SERVER_PORT"
+
+# Default FTS port
+DEFAULT_FEATURE_TRANSFORMATION_SERVER_PORT = 6569

sdk/python/feast/infra/transformation_servers/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.7-slim
+
+# Copy app handler code
+COPY sdk/python/feast/infra/transformation_servers/app.py app.py
+
+# Copy necessary parts of the Feast codebase
+COPY sdk/python sdk/python
+COPY protos protos
+COPY README.md README.md
+
+# Install dependencies
+RUN pip3 install -e 'sdk/python[ci]'
+
+# Start feature transformation server
+CMD [ "python", "app.py" ]

sdk/python/feast/infra/transformation_servers/app.py

@@ -0,0 +1,50 @@
+import base64
+import os
+import tempfile
+from pathlib import Path
+
+import yaml
+
+from feast import FeatureStore
+from feast.constants import (
+    DEFAULT_FEATURE_TRANSFORMATION_SERVER_PORT,
+    FEATURE_STORE_YAML_ENV_NAME,
+    FEATURE_TRANSFORMATION_SERVER_PORT_ENV_NAME,
+    REGISTRY_ENV_NAME,
+)
+from feast.infra.local import LocalRegistryStore
+from feast.registry import get_registry_store_class_from_scheme
+
+# Load RepoConfig
+config_base64 = os.environ[FEATURE_STORE_YAML_ENV_NAME]
+config_bytes = base64.b64decode(config_base64)
+
+# Create a new unique directory for writing feature_store.yaml
+repo_path = Path(tempfile.mkdtemp())
+
+with open(repo_path / "feature_store.yaml", "wb") as f:
+    f.write(config_bytes)
+
+# Write registry contents for local registries
+config_string = config_bytes.decode("utf-8")
+raw_config = yaml.safe_load(config_string)
+registry_path = raw_config["registry"]
+registry_store_class = get_registry_store_class_from_scheme(registry_path)
+if registry_store_class == LocalRegistryStore:
+    registry_base64 = os.environ[REGISTRY_ENV_NAME]
+    registry_bytes = base64.b64decode(registry_base64)
+    registry_dir = os.path.dirname(registry_path)
+    if not os.path.exists(repo_path / registry_dir):
+        os.makedirs(repo_path / registry_dir)
+    with open(repo_path / registry_path, "wb") as f:
+        f.write(registry_bytes)
+
+# Initialize the feature store
+store = FeatureStore(repo_path=str(repo_path.resolve()))
+
+# Start the feature transformation server
+port = (
+    os.environ.get(FEATURE_TRANSFORMATION_SERVER_PORT_ENV_NAME)
+    or DEFAULT_FEATURE_TRANSFORMATION_SERVER_PORT
+)
+store.serve_transformations(port)