Visitar URL original
fix: Provision online store infra for remote feast apply by RishabhCodezZz · Pull Request #6772 · feast-dev/feast · GitHub
Skip to content
Open
Show file tree
Hide file tree
Changes from 5 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])
Comment on lines +1032 to +1035
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])

Comment on lines +1061 to +1065
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)}"
)
Comment on lines +112 to +117
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
Loading