fix: Provision online store infra for remote feast apply

docs/reference/feature-servers/python-feature-server.md

@@ -665,7 +665,19 @@ The [PyTorch NLP template](https://github.com/feast-dev/feast/tree/main/sdk/pyth
 | /write-to-online-store     | FeatureView                     | Write Online                                          | Write features to the online store                             |
 | /materialize               | FeatureView                     | Write Online                                          | Materialize features within a specified time range             |
 | /materialize-incremental   | FeatureView                     | Write Online                                          | Incrementally materialize features up to a specified timestamp |
+| /update-infra              | FeatureView,Entity              | Create, Delete                                        | Provision online store infrastructure for a remote `feast apply` |
+| /teardown-infra            | Project,FeatureView,Entity      | Delete                                                | Drop online store infrastructure for a remote `feast teardown`   |
 
 ## How to configure Authentication and Authorization ?
 
-Please refer the [page](./../../../docs/getting-started/concepts/permission.md) for more details on how to configure authentication and authorization.
+Remote `feast apply` requires a Python feature server that supports `/update-infra`.
+The client and server must use the same project; provisioning requests for another project are rejected.
+The provisioning endpoints manage the server's online store; compute-engine infrastructure remains with the client's provider.
+Applying or reapplying a view requires `Create` permission on its feature views and entities, in addition to any registry `Describe`/`Update` permissions.
+Full teardown also requires `Delete` permission on the server's project because some online stores remove shared infrastructure even when the request lists are empty.
+When a remote registry has already removed a deletion target, `/update-infra` requires project-level `Delete` permission to clean up its remaining infrastructure.
+Without that permission, the registry deletion can succeed while infrastructure cleanup fails; grant the cleanup actor project-level `Delete` before running the operation.
+Resource permissions are checked before decoding source implementations. Authorized provisioning requests can load custom data-source classes installed on the server, so grant provisioning permissions only to trusted infrastructure operators.
+Object permissions do not guarantee isolation between feature views that share a physical table in the configured online store.
+
+Please refer the [page](./../../../docs/getting-started/concepts/permission.md) for more details on how to configure authentication and authorization.

sdk/python/feast/feature_server.py

@@ -25,13 +25,14 @@
 from datetime import datetime
 from importlib import resources as importlib_resources
 from types import SimpleNamespace
-from typing import Any, DefaultDict, Dict, List, NamedTuple, Optional, Set, Union
+from typing import Any, DefaultDict, Dict, List, NamedTuple, Optional, Set, Union, cast
 
 import pandas as pd
 from dateutil import parser
 from fastapi import (
     Depends,
     FastAPI,
+    HTTPException,
     Query,
     Request,
     Response,
@@ -43,18 +44,22 @@
 from fastapi.logger import logger
 from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
+from google.protobuf.message import DecodeError
 from pydantic import BaseModel, Field, field_validator
+from typeguard import TypeCheckError
 
 import feast
 from feast import metrics as feast_metrics
 from feast import proto_json, utils
-from feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL
+from feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL, PROTECTED_PROJECT_TAG
 from feast.data_source import PushMode
 from feast.errors import (
     FeastError,
+    FeastObjectNotFoundException,
     FeatureViewNotFoundException,
+    ProjectNotFoundException,
 )
-from feast.feast_object import FeastObject
+from feast.feast_object import ALL_FEATURE_VIEW_TYPES, FeastObject
 from feast.feature_server_utils import convert_response_to_dict
 from feast.feature_view import FeatureViewState
 from feast.feature_view_utils import get_feature_view_from_feature_store
@@ -84,6 +89,23 @@ class WriteToFeatureStoreRequest(BaseModel):
     transform_on_write: bool = True
 
 
+class UpdateInfraRequest(BaseModel):
+    """Base64 protos carried before the client's registry commit."""
+
+    tables_to_delete: List[Dict[str, str]] = []
+    tables_to_keep: List[Dict[str, str]] = []
+    entities_to_delete: List[Dict[str, str]] = []
+    entities_to_keep: List[Dict[str, str]] = []
+    partial: bool = False
+
+
+class TeardownInfraRequest(BaseModel):
+    """Objects to tear down, sent by RemoteOnlineStore.teardown()."""
+
+    tables: List[Dict[str, str]] = []
+    entities: List[Dict[str, str]] = []
+
+
 class PushFeaturesRequest(BaseModel):
     push_source_name: str
     df: dict
@@ -987,6 +1009,107 @@ async def write_to_online_store(request: WriteToFeatureStoreRequest) -> None:
             transform_on_write=request.transform_on_write,
         )
 
+    @app.post("/update-infra", dependencies=[Depends(inject_user_details)])
+    async def update_infra(
+        request: UpdateInfraRequest, project: Optional[str] = Query(None)
+    ) -> None:
+        if project is not None and project != store.project:
+            raise HTTPException(400, "Infrastructure belongs to the server project")
+        tables_to_keep = _decode_authorized(
+            request.tables_to_keep, AuthzedAction.CREATE
+        )
+        tables_to_delete = _decode_authorized(
+            request.tables_to_delete, AuthzedAction.DELETE
+        )
+        entities_to_keep = _decode_authorized(
+            request.entities_to_keep, AuthzedAction.CREATE, entities=True
+        )
+        entities_to_delete = _decode_authorized(
+            request.entities_to_delete, AuthzedAction.DELETE, entities=True
+        )
+        if not any(
+            (tables_to_keep, tables_to_delete, entities_to_keep, entities_to_delete)
+        ):
+            return
+        await run_in_threadpool(
+            store._get_provider().online_store.update,
+            config=store.config,
+            tables_to_delete=tables_to_delete,
+            tables_to_keep=tables_to_keep,
+            entities_to_delete=entities_to_delete,
+            entities_to_keep=entities_to_keep,
+            partial=request.partial,
+        )
+
+    @app.post("/teardown-infra", dependencies=[Depends(inject_user_details)])
+    async def teardown_infra(
+        request: TeardownInfraRequest, project: Optional[str] = Query(None)
+    ) -> None:
+        if project is not None and project != store.project:
+            raise HTTPException(400, "Infrastructure belongs to the server project")
+        try:
+            server_project = store.registry.get_project(
+                store.project, allow_cache=False
+            )
+        except ProjectNotFoundException:
+            server_project = feast.Project(name=store.project)
+        # Some backends remove shared infrastructure even when the lists are empty.
+        assert_permissions(resource=server_project, actions=[AuthzedAction.DELETE])
+        if server_project.tags.get(PROTECTED_PROJECT_TAG) == "true":
+            raise HTTPException(403, "Cannot tear down a protected project")
+        tables = _decode_authorized(request.tables, AuthzedAction.DELETE)
+        entities = _decode_authorized(
+            request.entities, AuthzedAction.DELETE, entities=True
+        )
+        await run_in_threadpool(
+            store._get_provider().online_store.teardown,
+            config=store.config,
+            tables=tables,
+            entities=entities,
+        )
+
+    def _decode_authorized(
+        payloads: List[Dict[str, str]], action: AuthzedAction, *, entities: bool = False
+    ) -> List[Any]:
+        from feast.infra.online_stores.remote import decode_infra_object
+
+        def authorize(obj: Any) -> Optional[Any]:
+            if not isinstance(
+                obj, feast.Entity if entities else tuple(ALL_FEATURE_VIEW_TYPES)
+            ):
+                raise HTTPException(400, "Unexpected infrastructure resource type")
+            resource = obj
+            replacement = None
+            if is_auth_necessary(get_security_manager()):
+                getter = (
+                    store.registry.get_entity
+                    if entities
+                    else store.registry.get_any_feature_view
+                )
+                try:
+                    existing = cast(
+                        FeastObject, getter(obj.name, store.project, allow_cache=False)
+                    )
+                except FeastObjectNotFoundException:
+                    # A remote registry may already have committed the deletion.
+                    if action == AuthzedAction.DELETE:
+                        resource = feast.Project(name=store.project)
+                else:
+                    if action == AuthzedAction.DELETE:
+                        replacement = resource = existing
+                    else:
+                        assert_permissions(resource=existing, actions=[action])
+            assert_permissions(resource=resource, actions=[action])
+            return replacement
+
+        objects = []
+        for payload in payloads:
+            try:
+                objects.append(decode_infra_object(payload, authorize=authorize))
+            except (ValueError, KeyError, DecodeError, TypeCheckError) as exc:
+                raise HTTPException(400, str(exc)) from exc
+        return objects
+
     @app.get("/health")
     async def health():
         try:

sdk/python/feast/infra/online_stores/remote.py

@@ -11,6 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import base64
 import json
 import logging
 import uuid as uuid_module
@@ -33,13 +34,25 @@
 from pydantic import StrictStr
 
 from feast import Entity, FeatureView, RepoConfig
+from feast.batch_feature_view import BatchFeatureView
 from feast.feature_service import FeatureService
 from feast.filter_models import ComparisonFilter, CompoundFilter
 from feast.infra.online_stores.helpers import _to_naive_utc
 from feast.infra.online_stores.online_store import OnlineStore
 from feast.infra.registry.base_registry import BaseRegistry
+from feast.labeling.label_view import LabelView
+from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.online_response import OnlineResponse
 from feast.permissions.client.http_auth_requests_wrapper import HttpSessionManager
+from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto
+from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
+from feast.protos.feast.core.LabelView_pb2 import LabelView as LabelViewProto
+from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
+    OnDemandFeatureView as OnDemandFeatureViewProto,
+)
+from feast.protos.feast.core.StreamFeatureView_pb2 import (
+    StreamFeatureView as StreamFeatureViewProto,
+)
 from feast.protos.feast.serving.ServingService_pb2 import (
     FieldStatus,
     GetOnlineFeaturesResponse,
@@ -50,6 +63,7 @@
 from feast.protos.feast.types.Value_pb2 import Value as ValueProto
 from feast.repo_config import FeastConfigBaseModel
 from feast.rest_error_handler import rest_error_handling_decorator
+from feast.stream_feature_view import StreamFeatureView
 from feast.type_map import (
     feast_value_type_to_python_type,
     python_values_to_proto_values,
@@ -60,6 +74,69 @@
 logger = logging.getLogger(__name__)
 
 
+# Never execute UDFs from client payloads: provisioning only needs metadata.
+_INFRA_OBJECT_TYPES: Dict[str, Tuple[Any, Any, dict]] = {
+    "FeatureView": (FeatureView, FeatureViewProto, {"skip_udf": True}),
+    "BatchFeatureView": (FeatureView, FeatureViewProto, {"skip_udf": True}),
+    "StreamFeatureView": (
+        StreamFeatureView,
+        StreamFeatureViewProto,
+        {"skip_udf": True},
+    ),
+    "OnDemandFeatureView": (
+        OnDemandFeatureView,
+        OnDemandFeatureViewProto,
+        {"skip_udf": True},
+    ),
+    "LabelView": (LabelView, LabelViewProto, {}),
+    "Entity": (Entity, EntityProto, {}),
+}
+
+
+def encode_infra_object(obj: Any) -> Dict[str, str]:
+    """Send objects in the body because apply provisions before registry.commit()."""
+    type_name = type(obj).__name__
+    if type_name not in _INFRA_OBJECT_TYPES:
+        raise ValueError(
+            f"Cannot send {type_name} to the remote online store; "
+            f"expected one of {sorted(_INFRA_OBJECT_TYPES)}"
+        )
+    return {
+        "type": type_name,
+        "proto": base64.b64encode(obj.to_proto().SerializeToString()).decode("ascii"),
+    }
+
+
+def decode_infra_object(
+    payload: Mapping[str, str],
+    *,
+    authorize: Optional[Callable[[Any], Optional[Any]]] = None,
+) -> Any:
+    """Authorize source-free identity before reconstructing allowlisted metadata."""
+    type_name = payload["type"]
+    if type_name not in _INFRA_OBJECT_TYPES:
+        raise ValueError(
+            f"Unknown object type {type_name!r}; expected one of {sorted(_INFRA_OBJECT_TYPES)}"
+        )
+    feast_class, proto_class, kwargs = _INFRA_OBJECT_TYPES[type_name]
+    proto = proto_class.FromString(base64.b64decode(payload["proto"], validate=True))
+    if authorize is not None:
+        resource_class = feast_class
+        if isinstance(proto, FeatureViewProto) and proto.spec.HasField(
+            "feature_transformation"
+        ):
+            resource_class = BatchFeatureView
+        # Permission matching only needs type/name/tags; constructors can import sources.
+        identity = object.__new__(resource_class)
+        identity.name = proto.spec.name
+        identity.tags = dict(proto.spec.tags)
+        identity.required_tags = identity.tags
+        replacement = authorize(identity)
+        if replacement is not None:
+            return replacement
+    return feast_class.from_proto(proto, **kwargs)
+
+
 def _json_safe(val: Any) -> Any:
     """Convert uuid.UUID objects and sets to JSON-serializable form."""
     if isinstance(val, uuid_module.UUID):
@@ -741,16 +818,37 @@ def update(
         entities_to_delete: Sequence[Entity],
         entities_to_keep: Sequence[Entity],
         partial: bool,
-    ):
-        pass
+    ) -> None:
+        """Provision the server's concrete online store."""
+        post_remote_infra(
+            config=config,
+            endpoint="update-infra",
+            req_body={
+                "tables_to_delete": [encode_infra_object(t) for t in tables_to_delete],
+                "tables_to_keep": [encode_infra_object(t) for t in tables_to_keep],
+                "entities_to_delete": [
+                    encode_infra_object(e) for e in entities_to_delete
+                ],
+                "entities_to_keep": [encode_infra_object(e) for e in entities_to_keep],
+                "partial": partial,
+            },
+        )
 
     def teardown(
         self,
         config: RepoConfig,
         tables: Sequence[FeatureView],
         entities: Sequence[Entity],
-    ):
-        pass
+    ) -> None:
+        """Tear down the server's online infrastructure."""
+        post_remote_infra(
+            config=config,
+            endpoint="teardown-infra",
+            req_body={
+                "tables": [encode_infra_object(t) for t in tables],
+                "entities": [encode_infra_object(e) for e in entities],
+            },
+        )
 
     async def close(self) -> None:
         """
@@ -812,3 +910,22 @@ def post_remote_online_write(
         return session.post(url, json=req_body, verify=config.online_store.cert)
     else:
         return session.post(url, json=req_body)
+
+
+@rest_error_handling_decorator
+def post_remote_infra(
+    session: requests.Session, config: RepoConfig, endpoint: str, req_body: dict
+) -> None:
+    """Send a provisioning request with the existing authenticated session."""
+    assert isinstance(config.online_store, RemoteOnlineStoreConfig)
+    response = session.post(
+        f"{config.online_store.path.rstrip('/')}/{endpoint}",
+        params={"project": config.project},
+        json=req_body,
+        verify=config.online_store.cert or True,
+    )
+    if response.status_code != 200:
+        raise RuntimeError(
+            f"Unable to {endpoint.split('-')[0]} online store infrastructure. "
+            f"Error_code={response.status_code}, error_message={response.text}"
+        )

sdk/python/tests/integration/online_store/test_remote_online_store.py

@@ -291,6 +291,12 @@ def test_remote_online_store_read_write(auth_config, tls_mode):
         tempfile.TemporaryDirectory() as remote_client_tmp_dir,
     ):
         permissions_list = [
+            Permission(
+                name="online_provision_perm",
+                types=[FeatureView, Entity],
+                policy=RoleBasedPolicy(roles=["writer"]),
+                actions=[AuthzedAction.CREATE],
+            ),
             Permission(
                 name="online_list_fv_perm",
                 types=FeatureView,