feat: Add Feast Operator RBAC example with Kubernetes Authentication …

…type.

Signed-off-by: Abdul Hameed <[email protected]>

examples/README.md

@@ -18,4 +18,5 @@ The following examples illustrate various **Feast** use cases to enhance underst
 The examples below showcase how to deploy and manage **Feast on Kubernetes** using the **Feast Go Operator**.
 
 1. **[Operator Quickstart](operator-quickstart)**: Demonstrates how to install and use Feast on Kubernetes with the Feast Go Operator.
-1. **[Operator Quickstart with Postgres in TLS](operator-postgres-tls-demo)**: Demonstrates installing and configuring Feast with PostgreSQL in TLS mode on Kubernetes using the Feast Go Operator, with an emphasis on volumes and VolumeMounts support.  
+1. **[Operator Quickstart with Postgres in TLS](operator-postgres-tls-demo)**: Demonstrates installing and configuring Feast with PostgreSQL in TLS mode on Kubernetes using the Feast Go Operator, with an emphasis on volumes and VolumeMounts support.
+1. **[Operator RBAC with Kubernetes](operator-rbac)**: Demonstrates the Feast RBAC example on Kubernetes using the Feast Operator.

examples/operator-rbac/README.md

@@ -0,0 +1,6 @@
+# Running the Feast RBAC example on Kubernetes using the Feast Operator.
+
+The [k8s-rbac.ipynb](k8s-rbac.ipynb) will guide you through how to enable Role-Based Access Control (RBAC) for Feast using [Feast Operator](../../infra/feast-operator/)  with the Kubernetes Authentication type.
+
+
+

examples/operator-rbac/client/feature_store.yaml

@@ -0,0 +1,15 @@
+project: feast_rbac
+provider: local
+offline_store:
+    host: localhost
+    type: remote
+    port: 8081
+online_store:
+    path: http://localhost:8082
+    type: remote
+registry:
+    path: localhost:8083
+    registry_type: remote
+auth:
+    type: kubernetes
+entity_key_serialization_version: 3

examples/operator-rbac/client/test.py

@@ -0,0 +1,139 @@
+import os
+from datetime import datetime
+
+import pandas as pd
+from feast import FeatureStore
+from feast.data_source import PushMode
+
+
+def run_demo():
+    try:
+
+        store = FeatureStore(repo_path="client")
+
+        print("\n--- Historical features for training ---")
+        fetch_historical_features_entity_df(store, for_batch_scoring=False)
+
+        print("\n--- Historical features for batch scoring ---")
+        fetch_historical_features_entity_df(store, for_batch_scoring=True)
+
+        try:
+            print("\n--- Load features into online store/materialize_incremental ---")
+            feature_views= store.list_feature_views()
+            if not feature_views:
+                raise PermissionError("No access to feature-views or no feature-views available.")
+            store.materialize_incremental(end_date=datetime.now())
+        except PermissionError as pe:
+            print(f"Permission error: {pe}")
+        except Exception as e:
+            print(f"An occurred while performing materialize incremental: {e}")
+
+        print("\n--- Online features ---")
+        fetch_online_features(store)
+
+        print("\n--- Online features retrieved (instead) through a feature service---")
+        fetch_online_features(store, source="feature_service")
+
+        print(
+            "\n--- Online features retrieved (using feature service v3, which uses a feature view with a push source---"
+        )
+        fetch_online_features(store, source="push")
+
+        print("\n--- Simulate a stream event ingestion of the hourly stats df ---")
+        event_df = pd.DataFrame.from_dict(
+            {
+                "driver_id": [1001],
+                "event_timestamp": [datetime.now()],
+                "created": [datetime.now()],
+                "conv_rate": [1.0],
+                "acc_rate": [1.0],
+                "avg_daily_trips": [1000],
+            }
+        )
+        store.push("driver_stats_push_source", event_df, to=PushMode.ONLINE_AND_OFFLINE)
+
+        print("\n--- Online features again with updated values from a stream push---")
+        fetch_online_features(store, source="push")
+
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+def fetch_historical_features_entity_df(store: FeatureStore, for_batch_scoring: bool):
+    try:
+        entity_df = pd.DataFrame.from_dict(
+            {
+                "driver_id": [1001, 1002, 1003],
+                "event_timestamp": [
+                    datetime(2021, 4, 12, 10, 59, 42),
+                    datetime(2021, 4, 12, 8, 12, 10),
+                    datetime(2021, 4, 12, 16, 40, 26),
+                ],
+                "label_driver_reported_satisfaction": [1, 5, 3],
+                # values we're using for an on-demand transformation
+                "val_to_add": [1, 2, 3],
+                "val_to_add_2": [10, 20, 30],
+
+            }
+
+        )
+        if for_batch_scoring:
+            entity_df["event_timestamp"] = pd.to_datetime("now", utc=True)
+
+        training_df = store.get_historical_features(
+            entity_df=entity_df,
+            features=[
+                "driver_hourly_stats:conv_rate",
+                "driver_hourly_stats:acc_rate",
+                "driver_hourly_stats:avg_daily_trips",
+                "transformed_conv_rate:conv_rate_plus_val1",
+                "transformed_conv_rate:conv_rate_plus_val2",
+            ],
+        ).to_df()
+        print(training_df.head())
+
+    except Exception as e:
+        print(f"An error occurred while fetching historical features: {e}")
+
+
+def fetch_online_features(store, source: str = ""):
+    try:
+        entity_rows = [
+            # {join_key: entity_value}
+            {
+                "driver_id": 1001,
+                "val_to_add": 1000,
+                "val_to_add_2": 2000,
+            },
+            {
+                "driver_id": 1002,
+                "val_to_add": 1001,
+                "val_to_add_2": 2002,
+            },
+        ]
+        if source == "feature_service":
+            features_to_fetch = store.get_feature_service("driver_activity_v1")
+        elif source == "push":
+            features_to_fetch = store.get_feature_service("driver_activity_v3")
+        else:
+            features_to_fetch = [
+                "driver_hourly_stats:acc_rate",
+                "transformed_conv_rate:conv_rate_plus_val1",
+                "transformed_conv_rate:conv_rate_plus_val2",
+            ]
+        returned_features = store.get_online_features(
+            features=features_to_fetch,
+            entity_rows=entity_rows,
+        ).to_dict()
+        for key, value in sorted(returned_features.items()):
+            print(key, " : ", value)
+
+    except Exception as e:
+        print(f"An error occurred while fetching online features: {e}")
+
+
+if __name__ == "__main__":
+    try:
+        run_demo()
+    except Exception as e:
+        print(f"An error occurred in the main execution: {e}")