feat: Added odfv transformations metrics

Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>

Author: ntkathole

docs/getting-started/components/open-telemetry.md

@@ -144,6 +144,12 @@ Once configured, you can monitor various metrics including:
 
 - `feast_feature_server_memory_usage`: Memory utilization of the feature server
 - `feast_feature_server_cpu_usage`: CPU usage statistics
+- `feast_feature_server_request_latency_seconds`: Request latency with feature count dimensions
+- `feast_feature_server_online_store_read_duration_seconds`: Online store read phase duration
+- `feast_feature_server_transformation_duration_seconds`: ODFV read-path transformation duration (per ODFV, requires `track_metrics=True`)
+- `feast_feature_server_write_transformation_duration_seconds`: ODFV write-path transformation duration (per ODFV, requires `track_metrics=True`)
 - Additional custom metrics based on your configuration
 
+For the full list of metrics, see the [Python Feature Server reference](../../reference/feature-servers/python-feature-server.md#available-metrics).
+
 These metrics can be visualized using Prometheus and other compatible monitoring tools.

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

@@ -360,18 +360,50 @@ thread from starting). All categories default to `true`.
 
 ### Available metrics
 
-| Metric | Type | Labels | Description |
-|--------|------|--------|-------------|
-| `feast_feature_server_cpu_usage` | Gauge | — | Process CPU usage % |
-| `feast_feature_server_memory_usage` | Gauge | — | Process memory usage % |
-| `feast_feature_server_request_total` | Counter | `endpoint`, `status` | Total requests per endpoint |
-| `feast_feature_server_request_latency_seconds` | Histogram | `endpoint`, `feature_count`, `feature_view_count` | Request latency with p50/p95/p99 support |
-| `feast_online_features_request_total` | Counter | — | Total online feature retrieval requests |
-| `feast_online_features_entity_count` | Histogram | — | Entity rows per online feature request |
-| `feast_push_request_total` | Counter | `push_source`, `mode` | Push requests by source and mode |
-| `feast_materialization_total` | Counter | `feature_view`, `status` | Materialization runs (success/failure) |
-| `feast_materialization_duration_seconds` | Histogram | `feature_view` | Materialization duration per feature view |
-| `feast_feature_freshness_seconds` | Gauge | `feature_view`, `project` | Seconds since last materialization |
+| Metric | Type | Labels | Category | Description |
+|--------|------|--------|----------|-------------|
+| `feast_feature_server_cpu_usage` | Gauge | — | `resource` | Process CPU usage % |
+| `feast_feature_server_memory_usage` | Gauge | — | `resource` | Process memory usage % |
+| `feast_feature_server_request_total` | Counter | `endpoint`, `status` | `request` | Total requests per endpoint |
+| `feast_feature_server_request_latency_seconds` | Histogram | `endpoint`, `feature_count`, `feature_view_count` | `request` | Request latency with p50/p95/p99 support |
+| `feast_online_features_request_total` | Counter | — | `online_features` | Total online feature retrieval requests |
+| `feast_online_features_entity_count` | Histogram | — | `online_features` | Entity rows per online feature request |
+| `feast_feature_server_online_store_read_duration_seconds` | Histogram | — | `online_features` | Online store read phase duration (sync and async) |
+| `feast_feature_server_transformation_duration_seconds` | Histogram | `odfv_name`, `mode` | `online_features` | ODFV read-path transformation duration (requires `track_metrics=True` on the ODFV) |
+| `feast_feature_server_write_transformation_duration_seconds` | Histogram | `odfv_name`, `mode` | `online_features` | ODFV write-path transformation duration (requires `track_metrics=True` on the ODFV) |
+| `feast_push_request_total` | Counter | `push_source`, `mode` | `push` | Push requests by source and mode |
+| `feast_materialization_result_total` | Counter | `feature_view`, `status` | `materialization` | Materialization runs (success/failure) |
+| `feast_materialization_duration_seconds` | Histogram | `feature_view` | `materialization` | Materialization duration per feature view |
+| `feast_feature_freshness_seconds` | Gauge | `feature_view`, `project` | `freshness` | Seconds since last materialization |
+
+### Per-ODFV transformation metrics
+
+The `transformation_duration_seconds` and `write_transformation_duration_seconds`
+metrics are gated behind **two** conditions — both must be true for any
+instrumentation to run:
+
+1. **Server-level**: the `online_features` category must be enabled in the
+   metrics configuration.
+2. **ODFV-level**: the `OnDemandFeatureView` must have `track_metrics=True`.
+
+This defaults to `False`, so no ODFV incurs timing overhead unless explicitly
+opted in:
+
+```python
+from feast.on_demand_feature_view import on_demand_feature_view
+
+@on_demand_feature_view(
+    sources=[my_feature_view, my_request_source],
+    schema=[Field(name="output", dtype=Float64)],
+    track_metrics=True,   # opt in to transformation timing
+)
+def my_transform(inputs: pd.DataFrame) -> pd.DataFrame:
+    ...
+```
+
+The `odfv_name` label lets you filter or group by individual ODFV,
+and the `mode` label (`python`, `pandas`, `substrait`) lets you compare
+transformation engines.
 
 ### Scraping with Prometheus
 

docs/reference/feature-store-yaml.md

@@ -40,7 +40,7 @@ feature_server:
     enabled: true             # Enable Prometheus metrics server on port 8000
     resource: true            # CPU / memory gauges
     request: true             # endpoint latency histograms & request counters
-    online_features: true     # online feature retrieval counters
+    online_features: true     # online feature retrieval counters + store read & ODFV transform timing
     push: true                # push request counters
     materialization: true     # materialization counters & duration histograms
     freshness: true           # per-feature-view freshness gauges

sdk/python/feast/feature_store.py

@@ -2097,76 +2097,104 @@ def _transform_on_demand_feature_view_df(
         Raises:
             Exception: For unsupported OnDemandFeatureView modes
         """
-        if feature_view.mode == "python" and isinstance(
-            feature_view.feature_transformation, PythonTransformation
-        ):
-            input_dict = (
-                df.to_dict(orient="records")[0]
-                if feature_view.singleton
-                else df.to_dict(orient="list")
+        _should_track = False
+        try:
+            from feast.metrics import _config as _metrics_config
+
+            _should_track = _metrics_config.online_features and getattr(
+                feature_view, "track_metrics", False
             )
+        except Exception:
+            pass
 
-            if feature_view.singleton:
-                transformed_rows = []
+        if _should_track:
+            import time as _time
 
-                for i, row in df.iterrows():
-                    output = feature_view.feature_transformation.udf(row.to_dict())
-                    if i == 0:
-                        transformed_rows = output
-                    else:
-                        for k in output:
-                            if isinstance(output[k], list):
-                                transformed_rows[k].extend(output[k])
-                            else:
-                                transformed_rows[k].append(output[k])
-
-                transformed_data = pd.DataFrame(transformed_rows)
-            else:
-                transformed_data = feature_view.feature_transformation.udf(input_dict)
+            _t0 = _time.monotonic()
 
-            if feature_view.write_to_online_store:
-                entities = [
-                    self.get_entity(entity) for entity in (feature_view.entities or [])
-                ]
-                join_keys = [entity.join_key for entity in entities if entity]
-                join_keys = [k for k in join_keys if k in input_dict.keys()]
-                transformed_df = (
-                    pd.DataFrame(transformed_data)
-                    if not isinstance(transformed_data, pd.DataFrame)
-                    else transformed_data
-                )
-                input_df = pd.DataFrame(
-                    [input_dict] if feature_view.singleton else input_dict
+        try:
+            if feature_view.mode == "python" and isinstance(
+                feature_view.feature_transformation, PythonTransformation
+            ):
+                input_dict = (
+                    df.to_dict(orient="records")[0]
+                    if feature_view.singleton
+                    else df.to_dict(orient="list")
                 )
-                if input_df.shape[0] == transformed_df.shape[0]:
+
+                if feature_view.singleton:
+                    transformed_rows = []
+
+                    for i, row in df.iterrows():
+                        output = feature_view.feature_transformation.udf(row.to_dict())
+                        if i == 0:
+                            transformed_rows = output
+                        else:
+                            for k in output:
+                                if isinstance(output[k], list):
+                                    transformed_rows[k].extend(output[k])
+                                else:
+                                    transformed_rows[k].append(output[k])
+
+                    transformed_data = pd.DataFrame(transformed_rows)
+                else:
+                    transformed_data = feature_view.feature_transformation.udf(
+                        input_dict
+                    )
+
+                if feature_view.write_to_online_store:
+                    entities = [
+                        self.get_entity(entity)
+                        for entity in (feature_view.entities or [])
+                    ]
+                    join_keys = [entity.join_key for entity in entities if entity]
+                    join_keys = [k for k in join_keys if k in input_dict.keys()]
+                    transformed_df = (
+                        pd.DataFrame(transformed_data)
+                        if not isinstance(transformed_data, pd.DataFrame)
+                        else transformed_data
+                    )
+                    input_df = pd.DataFrame(
+                        [input_dict] if feature_view.singleton else input_dict
+                    )
+                    if input_df.shape[0] == transformed_df.shape[0]:
+                        for k in input_dict:
+                            if k not in transformed_data:
+                                transformed_data[k] = input_dict[k]
+                        transformed_df = pd.DataFrame(transformed_data)
+                    else:
+                        transformed_df = pd.merge(
+                            transformed_df,
+                            input_df,
+                            how="left",
+                            on=join_keys,
+                        )
+                else:
+                    # overwrite any transformed features and update the dictionary
                     for k in input_dict:
                         if k not in transformed_data:
                             transformed_data[k] = input_dict[k]
-                    transformed_df = pd.DataFrame(transformed_data)
-                else:
-                    transformed_df = pd.merge(
-                        transformed_df,
-                        input_df,
-                        how="left",
-                        on=join_keys,
-                    )
-            else:
-                # overwrite any transformed features and update the dictionary
-                for k in input_dict:
-                    if k not in transformed_data:
-                        transformed_data[k] = input_dict[k]
 
-            return pd.DataFrame(transformed_data)
+                return pd.DataFrame(transformed_data)
 
-        elif feature_view.mode == "pandas" and isinstance(
-            feature_view.feature_transformation, PandasTransformation
-        ):
-            transformed_df = feature_view.feature_transformation.udf(df)
-            for col in df.columns:
-                transformed_df[col] = df[col]
-            return transformed_df
-        else:
-            raise Exception("Unsupported OnDemandFeatureView mode")
+            elif feature_view.mode == "pandas" and isinstance(
+                feature_view.feature_transformation, PandasTransformation
+            ):
+                transformed_df = feature_view.feature_transformation.udf(df)
+                for col in df.columns:
+                    transformed_df[col] = df[col]
+                return transformed_df
+            else:
+                raise Exception("Unsupported OnDemandFeatureView mode")
+        finally:
+            if _should_track:
+                from feast.metrics import track_write_transformation
+
+                track_write_transformation(
+                    feature_view.name,
+                    feature_view.mode,
+                    _time.monotonic() - _t0,
+                )
 
     def _validate_vector_features(self, feature_view, df: pd.DataFrame) -> None:
         """

sdk/python/feast/infra/feature_servers/base_config.py

@@ -62,15 +62,20 @@ class MetricsConfig(FeastConfigBaseModel):
     online_features: StrictBool = True
     """Emit online feature retrieval metrics
     (feast_online_features_request_total,
-    feast_online_features_entity_count)."""
+    feast_online_features_entity_count,
+    feast_feature_server_online_store_read_duration_seconds,
+    feast_feature_server_transformation_duration_seconds,
+    feast_feature_server_write_transformation_duration_seconds).
+    ODFV transformation metrics additionally require track_metrics=True
+    on the OnDemandFeatureView definition."""
 
     push: StrictBool = True
     """Emit push/write request counters
     (feast_push_request_total)."""
 
     materialization: StrictBool = True
     """Emit materialization success/failure counters and duration histograms
-    (feast_materialization_total,
+    (feast_materialization_result_total,
     feast_materialization_duration_seconds)."""
 
     freshness: StrictBool = True

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

@@ -185,6 +185,19 @@ def get_online_features(
             native_entity_values=True,
         )
 
+        _track_read = False
+        try:
+            from feast.metrics import _config as _metrics_config
+
+            _track_read = _metrics_config.online_features
+        except Exception:
+            pass
+
+        if _track_read:
+            import time as _time
+
+            _read_start = _time.monotonic()
+
         for table, requested_features in grouped_refs:
             # Get the correct set of entity values with the correct join keys.
             table_entity_values, idxs, output_len = utils._get_unique_entities(
@@ -218,6 +231,11 @@ def get_online_features(
                 output_len,
             )
 
+        if _track_read:
+            from feast.metrics import track_online_store_read
+
+            track_online_store_read(_time.monotonic() - _read_start)
+
         if requested_on_demand_feature_views:
             utils._augment_response_with_on_demand_transforms(
                 online_features_response,
@@ -293,6 +311,19 @@ async def query_table(table, requested_features):
 
             return idxs, read_rows, output_len
 
+        _track_read = False
+        try:
+            from feast.metrics import _config as _metrics_config
+
+            _track_read = _metrics_config.online_features
+        except Exception:
+            pass
+
+        if _track_read:
+            import time as _time
+
+            _read_start = _time.monotonic()
+
         all_responses = await asyncio.gather(
             *[
                 query_table(table, requested_features)
@@ -318,6 +349,11 @@ async def query_table(table, requested_features):
                 output_len,
             )
 
+        if _track_read:
+            from feast.metrics import track_online_store_read
+
+            track_online_store_read(_time.monotonic() - _read_start)
+
         if requested_on_demand_feature_views:
             utils._augment_response_with_on_demand_transforms(
                 online_features_response,