Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/reference/feature-servers/python-feature-server.md
Original file line number Diff line number Diff line change
Expand Up @@ -643,6 +643,8 @@ 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 | Create, Delete | Provision online store infrastructure for a remote `feast apply` |
| /teardown-infra | FeatureView | Delete | Drop online store infrastructure for a remote `feast teardown` |

## How to configure Authentication and Authorization ?

Expand Down
83 changes: 83 additions & 0 deletions sdk/python/feast/feature_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,29 @@ class WriteToFeatureStoreRequest(BaseModel):
transform_on_write: bool = True


class UpdateInfraRequest(BaseModel):
"""Objects to provision, sent by RemoteOnlineStore.update().

Each entry is `{"type": <class name>, "proto": <base64 proto>}`. They are
carried in the body rather than looked up in the registry because
`FeatureStore.apply()` runs `update_infra()` before `registry.commit()`,
so the server cannot see them yet.
"""

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
Expand Down Expand Up @@ -932,6 +955,66 @@ 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) -> None:
"""Provision online store infrastructure on behalf of a remote client.

`feast apply` against `online_store: type: remote` cannot create tables
itself -- the concrete online store lives here. Without this the feature
view is registered but its table never exists (#6693).
"""
# Imported here, not at module scope: feast.infra.online_stores.remote
# imports from `feast`, which is still initializing when this module loads.
from feast.infra.online_stores.remote import decode_infra_object

tables_to_keep = [decode_infra_object(o) for o in request.tables_to_keep]
tables_to_delete = [decode_infra_object(o) for o in request.tables_to_delete]
entities_to_keep = [decode_infra_object(o) for o in request.entities_to_keep]
entities_to_delete = [
decode_infra_object(o) for o in request.entities_to_delete
]

for table in tables_to_keep:
assert_permissions(resource=table, actions=[AuthzedAction.CREATE])
for table in tables_to_delete:
assert_permissions(resource=table, actions=[AuthzedAction.DELETE])
for entity in entities_to_keep:
assert_permissions(resource=entity, actions=[AuthzedAction.CREATE])
for entity in entities_to_delete:
assert_permissions(resource=entity, actions=[AuthzedAction.DELETE])

await run_in_threadpool(
store._get_provider().update_infra,
project=store.project,
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) -> None:
"""Drop online store infrastructure on behalf of a remote client."""
# Imported here, not at module scope: feast.infra.online_stores.remote
# imports from `feast`, which is still initializing when this module loads.
from feast.infra.online_stores.remote import decode_infra_object

tables = [decode_infra_object(o) for o in request.tables]
entities = [decode_infra_object(o) for o in request.entities]

for table in tables:
assert_permissions(resource=table, actions=[AuthzedAction.DELETE])
for entity in entities:
assert_permissions(resource=entity, actions=[AuthzedAction.DELETE])

await run_in_threadpool(
store._get_provider().teardown_infra,
project=store.project,
tables=tables,
entities=entities,
)

@app.get("/health")
async def health():
try:
Expand Down
139 changes: 137 additions & 2 deletions sdk/python/feast/infra/online_stores/remote.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,13 @@
# 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
from collections import defaultdict
from datetime import datetime
from functools import lru_cache
from typing import (
Any,
Callable,
Expand Down Expand Up @@ -60,6 +62,71 @@
logger = logging.getLogger(__name__)


@lru_cache(maxsize=1)
def _infra_object_types() -> Dict[str, Tuple[Any, Any]]:
"""Map type name -> (feast class, proto class) for objects sent to /update-infra.

Imported lazily: these modules import from `feast` at module scope, and
pulling them in at the top of this file would create an import cycle.
Cached because `update()` calls this once per object in four lists, and
re-running the imports each time is pure overhead.
"""
from feast.labeling.label_view import LabelView
from feast.on_demand_feature_view import OnDemandFeatureView
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.stream_feature_view import StreamFeatureView

return {
"FeatureView": (FeatureView, FeatureViewProto),
"StreamFeatureView": (StreamFeatureView, StreamFeatureViewProto),
"OnDemandFeatureView": (OnDemandFeatureView, OnDemandFeatureViewProto),
"LabelView": (LabelView, LabelViewProto),
"Entity": (Entity, EntityProto),
}


def encode_infra_object(obj: Any) -> Dict[str, str]:
"""Serialize a feature view or entity for transport to the feature server.

The registry is not a usable channel here: `FeatureStore.apply()` calls
`update_infra()` *before* `registry.commit()`, so the server cannot yet see
these objects. They travel in the request body instead.
"""
types = _infra_object_types()
type_name = type(obj).__name__
if type_name not in types:
raise ValueError(
f"Cannot send {type_name} to the remote online store; "
f"expected one of {sorted(types)}"
)
return {
"type": type_name,
"proto": base64.b64encode(obj.to_proto().SerializeToString()).decode("ascii"),
}


def decode_infra_object(payload: Mapping[str, str]) -> Any:
"""Inverse of :func:`encode_infra_object`, used by the feature server."""
type_name = payload["type"]
types = _infra_object_types()
if type_name not in types:
raise ValueError(
f"Unknown object type {type_name!r}; expected one of {sorted(types)}"
)
feast_class, proto_class = types[type_name]
return feast_class.from_proto(
proto_class.FromString(base64.b64decode(payload["proto"]))
)


def _json_safe(val: Any) -> Any:
"""Convert uuid.UUID objects and sets to JSON-serializable form."""
if isinstance(val, uuid_module.UUID):
Expand Down Expand Up @@ -742,15 +809,61 @@ def update(
entities_to_keep: Sequence[Entity],
partial: bool,
):
pass
"""Provision online store infrastructure through the feature server.

Table creation happens inside the concrete online store's `update()`.
In remote mode that store lives server-side, so `feast apply` has to ask
the feature server to run it -- otherwise the feature view is registered
but its table never exists, and materialization fails with a missing
table (#6693).
"""
assert isinstance(config.online_store, RemoteOnlineStoreConfig)
config.online_store.__class__ = RemoteOnlineStoreConfig

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,
}

response = post_remote_update_infra(config=config, req_body=req_body)
if response.status_code != 200:
error_msg = (
"Unable to update online store infrastructure using feature server API. "
f"Error_code={response.status_code}, error_message={response.text}"
)
logger.error(error_msg)
raise RuntimeError(error_msg)

def teardown(
self,
config: RepoConfig,
tables: Sequence[FeatureView],
entities: Sequence[Entity],
):
pass
"""Drop online store infrastructure through the feature server.

The mirror of :meth:`update`: without this, `feast teardown` in remote
mode leaves every table behind.
"""
assert isinstance(config.online_store, RemoteOnlineStoreConfig)
config.online_store.__class__ = RemoteOnlineStoreConfig

req_body = {
"tables": [encode_infra_object(t) for t in tables],
"entities": [encode_infra_object(e) for e in entities],
}

response = post_remote_teardown_infra(config=config, req_body=req_body)
if response.status_code != 200:
error_msg = (
"Unable to teardown online store infrastructure using feature server API. "
f"Error_code={response.status_code}, error_message={response.text}"
)
logger.error(error_msg)
raise RuntimeError(error_msg)

async def close(self) -> None:
"""
Expand Down Expand Up @@ -812,3 +925,25 @@ 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_update_infra(
session: requests.Session, config: RepoConfig, req_body: dict
) -> requests.Response:
url = f"{config.online_store.path}/update-infra"
if config.online_store.cert:
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_teardown_infra(
session: requests.Session, config: RepoConfig, req_body: dict
) -> requests.Response:
url = f"{config.online_store.path}/teardown-infra"
if config.online_store.cert:
return session.post(url, json=req_body, verify=config.online_store.cert)
else:
return session.post(url, json=req_body)
Loading