From 03dac6f33d197d4dce25c0dd451c66dbe4d9fc3c Mon Sep 17 00:00:00 2001 From: Shumin Z Date: Thu, 20 Aug 2026 22:01:16 -0700 Subject: [PATCH 1/2] feat(operator): Add tolerations and nodeSelector to the FeatureStore CR (#6741) Allow scheduling the operator-managed Deployment on tainted or specific nodes via spec.services.tolerations and spec.services.nodeSelector, alongside the existing affinity and topologySpreadConstraints fields. Signed-off-by: Shumin --- .secrets.baseline | 18 +- docs/how-to-guides/feast-on-kubernetes.md | 42 +++++ .../api/v1/featurestore_types.go | 9 + .../api/v1/zz_generated.deepcopy.go | 14 ++ .../api/v1alpha1/featurestore_types.go | 9 + .../api/v1alpha1/zz_generated.deepcopy.go | 14 ++ .../manifests/feast.dev_featurestores.yaml | 164 ++++++++++++++++++ .../crd/bases/feast.dev_featurestores.yaml | 164 ++++++++++++++++++ infra/feast-operator/dist/install.yaml | 164 ++++++++++++++++++ infra/feast-operator/docs/api/markdown/ref.md | 5 + .../internal/controller/services/services.go | 21 ++- .../controller/services/services_test.go | 89 ++++++++++ .../featurestore_v1alpha1_scheduling_test.go | 62 +++++++ 13 files changed, 765 insertions(+), 10 deletions(-) create mode 100644 infra/feast-operator/test/api/featurestore_v1alpha1_scheduling_test.go diff --git a/.secrets.baseline b/.secrets.baseline index 1e5670af27a..bdeaa2d9d75 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -957,7 +957,7 @@ "filename": "infra/feast-operator/api/v1/featurestore_types.go", "hashed_secret": "44e17306b837162269a410204daaa5ecee4ec22c", "is_verified": false, - "line_number": 1066 + "line_number": 1075 } ], "infra/feast-operator/api/v1/zz_generated.deepcopy.go": [ @@ -966,21 +966,21 @@ "filename": "infra/feast-operator/api/v1/zz_generated.deepcopy.go", "hashed_secret": "f914fc9324de1bec1ad13dec94a8ea2ddb41fc87", "is_verified": false, - "line_number": 939 + "line_number": 953 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/api/v1/zz_generated.deepcopy.go", "hashed_secret": "44e17306b837162269a410204daaa5ecee4ec22c", "is_verified": false, - "line_number": 1000 + "line_number": 1014 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/api/v1/zz_generated.deepcopy.go", "hashed_secret": "c2028031c154bbe86fd69bef740855c74b927dcf", "is_verified": false, - "line_number": 1717 + "line_number": 1731 } ], "infra/feast-operator/api/v1alpha1/featurestore_types.go": [ @@ -989,7 +989,7 @@ "filename": "infra/feast-operator/api/v1alpha1/featurestore_types.go", "hashed_secret": "44e17306b837162269a410204daaa5ecee4ec22c", "is_verified": false, - "line_number": 669 + "line_number": 678 } ], "infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go": [ @@ -998,21 +998,21 @@ "filename": "infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go", "hashed_secret": "f914fc9324de1bec1ad13dec94a8ea2ddb41fc87", "is_verified": false, - "line_number": 620 + "line_number": 634 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go", "hashed_secret": "44e17306b837162269a410204daaa5ecee4ec22c", "is_verified": false, - "line_number": 1128 + "line_number": 1142 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go", "hashed_secret": "c2028031c154bbe86fd69bef740855c74b927dcf", "is_verified": false, - "line_number": 1133 + "line_number": 1147 } ], "infra/feast-operator/bundle/manifests/openlineage-secret_v1_secret.yaml": [ @@ -1564,5 +1564,5 @@ } ] }, - "generated_at": "2026-08-20T03:02:13Z" + "generated_at": "2026-08-20T15:20:58Z" } diff --git a/docs/how-to-guides/feast-on-kubernetes.md b/docs/how-to-guides/feast-on-kubernetes.md index 923fc387032..cb989d642bf 100644 --- a/docs/how-to-guides/feast-on-kubernetes.md +++ b/docs/how-to-guides/feast-on-kubernetes.md @@ -74,6 +74,48 @@ batch jobs, and more via the operator, see the [Operator Configuration Guides](feast-operator/README.md). {% endhint %} +## Schedule FeatureStore pods on specific nodes + +Use `spec.services.nodeSelector` to require labels on the nodes that run a +FeatureStore. Use `spec.services.tolerations` to allow those pods onto nodes with +matching taints (node conditions that repel pods). A toleration permits a matching +taint; it does not select nodes by itself. + +For example, the following places the FeatureStore on Linux nodes dedicated to Feast +and permits the matching `dedicated=feast:NoSchedule` taint: + +```yaml +apiVersion: feast.dev/v1 +kind: FeatureStore +metadata: + name: sample +spec: + feastProject: my_project + services: + nodeSelector: + kubernetes.io/os: linux + workload: feast + tolerations: + - key: dedicated + operator: Equal + value: feast + effect: NoSchedule +``` + +These settings apply to the FeatureStore Deployment's pod, including its init +containers and every enabled Feast service container. A `nodeSelector` configured +under an individual service's `server` block is merged with the shared selector and +overrides `services.nodeSelector` for duplicate keys. Because enabled services share +one pod, use `services.nodeSelector` for the common placement policy; avoid conflicting +per-service selector values. + +To confirm the resolved placement after reconciliation: + +```sh +kubectl get deployment feast-sample -o jsonpath='{.spec.template.spec.nodeSelector}' +kubectl get deployment feast-sample -o jsonpath='{.spec.template.spec.tolerations}' +``` + ## Upgrading the Operator ### OLM-managed installations diff --git a/infra/feast-operator/api/v1/featurestore_types.go b/infra/feast-operator/api/v1/featurestore_types.go index dd41feaf51a..ef9b2f32356 100644 --- a/infra/feast-operator/api/v1/featurestore_types.go +++ b/infra/feast-operator/api/v1/featurestore_types.go @@ -536,6 +536,15 @@ type FeatureStoreServices struct { // pod anti-affinity rule to prefer spreading pods across nodes. // +optional Affinity *corev1.Affinity `json:"affinity,omitempty"` + // Tolerations are applied to the FeatureStore deployment pods, allowing them to + // be scheduled onto nodes with matching taints. + // +optional + Tolerations []corev1.Toleration `json:"tolerations,omitempty"` + // NodeSelector is a selector which must be true for the FeatureStore deployment + // pods to fit on a node. This selector must match a node's labels for the pod to + // be scheduled on that node. + // +optional + NodeSelector map[string]string `json:"nodeSelector,omitempty"` // ResourceClaims defines which ResourceClaims must be allocated // and reserved before the Pod is allowed to start. The resources // will be made available to those containers which consume them diff --git a/infra/feast-operator/api/v1/zz_generated.deepcopy.go b/infra/feast-operator/api/v1/zz_generated.deepcopy.go index cd7afefb2c8..bd7fe316118 100644 --- a/infra/feast-operator/api/v1/zz_generated.deepcopy.go +++ b/infra/feast-operator/api/v1/zz_generated.deepcopy.go @@ -460,6 +460,20 @@ func (in *FeatureStoreServices) DeepCopyInto(out *FeatureStoreServices) { *out = new(corev1.Affinity) (*in).DeepCopyInto(*out) } + if in.Tolerations != nil { + in, out := &in.Tolerations, &out.Tolerations + *out = make([]corev1.Toleration, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.NodeSelector != nil { + in, out := &in.NodeSelector, &out.NodeSelector + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } if in.ResourceClaims != nil { in, out := &in.ResourceClaims, &out.ResourceClaims *out = make([]corev1.PodResourceClaim, len(*in)) diff --git a/infra/feast-operator/api/v1alpha1/featurestore_types.go b/infra/feast-operator/api/v1alpha1/featurestore_types.go index ed35e2b6c76..f726ebb7178 100644 --- a/infra/feast-operator/api/v1alpha1/featurestore_types.go +++ b/infra/feast-operator/api/v1alpha1/featurestore_types.go @@ -305,6 +305,15 @@ type FeatureStoreServices struct { RunFeastApplyOnInit *bool `json:"runFeastApplyOnInit,omitempty"` // Volumes specifies the volumes to mount in the FeatureStore deployment. A corresponding `VolumeMount` should be added to whichever feast service(s) require access to said volume(s). Volumes []corev1.Volume `json:"volumes,omitempty"` + // Tolerations are applied to the FeatureStore deployment pods, allowing them to + // be scheduled onto nodes with matching taints. + // +optional + Tolerations []corev1.Toleration `json:"tolerations,omitempty"` + // NodeSelector is a selector which must be true for the FeatureStore deployment + // pods to fit on a node. This selector must match a node's labels for the pod to + // be scheduled on that node. + // +optional + NodeSelector map[string]string `json:"nodeSelector,omitempty"` } // OfflineStore configures the offline store service diff --git a/infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go b/infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go index 2345d07533a..6c2cdbfe795 100644 --- a/infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go +++ b/infra/feast-operator/api/v1alpha1/zz_generated.deepcopy.go @@ -352,6 +352,20 @@ func (in *FeatureStoreServices) DeepCopyInto(out *FeatureStoreServices) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.Tolerations != nil { + in, out := &in.Tolerations, &out.Tolerations + *out = make([]v1.Toleration, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + if in.NodeSelector != nil { + in, out := &in.NodeSelector, &out.NodeSelector + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FeatureStoreServices. diff --git a/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml b/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml index 858c11fbdd6..f2a879fbb33 100644 --- a/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml +++ b/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml @@ -2257,6 +2257,13 @@ spec: description: InitImage overrides the image for init containers (feast-init, feast-apply). type: string + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -4856,6 +4863,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration applies + to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration matches + to. + type: string + type: object + type: array topologySpreadConstraints: description: TopologySpreadConstraints defines how pods are spread across topology domains. @@ -9101,6 +9142,13 @@ spec: description: InitImage overrides the image for init containers (feast-init, feast-apply). type: string + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -11741,6 +11789,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration + applies to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration + matches to. + type: string + type: object + type: array topologySpreadConstraints: description: TopologySpreadConstraints defines how pods are spread across topology domains. @@ -14671,6 +14753,13 @@ spec: disableInitContainers: description: Disable the 'feast repo initialization' initContainer type: boolean + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -16524,6 +16613,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration applies + to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration matches + to. + type: string + type: object + type: array ui: description: Creates a UI server container properties: @@ -19196,6 +19319,13 @@ spec: disableInitContainers: description: Disable the 'feast repo initialization' initContainer type: boolean + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -21085,6 +21215,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration + applies to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration + matches to. + type: string + type: object + type: array ui: description: Creates a UI server container properties: diff --git a/infra/feast-operator/config/crd/bases/feast.dev_featurestores.yaml b/infra/feast-operator/config/crd/bases/feast.dev_featurestores.yaml index cc59339f0ad..5e1f21f69e6 100644 --- a/infra/feast-operator/config/crd/bases/feast.dev_featurestores.yaml +++ b/infra/feast-operator/config/crd/bases/feast.dev_featurestores.yaml @@ -2257,6 +2257,13 @@ spec: description: InitImage overrides the image for init containers (feast-init, feast-apply). type: string + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -4856,6 +4863,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration applies + to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration matches + to. + type: string + type: object + type: array topologySpreadConstraints: description: TopologySpreadConstraints defines how pods are spread across topology domains. @@ -9101,6 +9142,13 @@ spec: description: InitImage overrides the image for init containers (feast-init, feast-apply). type: string + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -11741,6 +11789,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration + applies to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration + matches to. + type: string + type: object + type: array topologySpreadConstraints: description: TopologySpreadConstraints defines how pods are spread across topology domains. @@ -14671,6 +14753,13 @@ spec: disableInitContainers: description: Disable the 'feast repo initialization' initContainer type: boolean + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -16524,6 +16613,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration applies + to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration matches + to. + type: string + type: object + type: array ui: description: Creates a UI server container properties: @@ -19196,6 +19319,13 @@ spec: disableInitContainers: description: Disable the 'feast repo initialization' initContainer type: boolean + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -21085,6 +21215,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration + applies to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration + matches to. + type: string + type: object + type: array ui: description: Creates a UI server container properties: diff --git a/infra/feast-operator/dist/install.yaml b/infra/feast-operator/dist/install.yaml index 9ac76fc46d1..7480b9423b5 100644 --- a/infra/feast-operator/dist/install.yaml +++ b/infra/feast-operator/dist/install.yaml @@ -2265,6 +2265,13 @@ spec: description: InitImage overrides the image for init containers (feast-init, feast-apply). type: string + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -4864,6 +4871,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration applies + to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration matches + to. + type: string + type: object + type: array topologySpreadConstraints: description: TopologySpreadConstraints defines how pods are spread across topology domains. @@ -9109,6 +9150,13 @@ spec: description: InitImage overrides the image for init containers (feast-init, feast-apply). type: string + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -11749,6 +11797,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration + applies to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration + matches to. + type: string + type: object + type: array topologySpreadConstraints: description: TopologySpreadConstraints defines how pods are spread across topology domains. @@ -14679,6 +14761,13 @@ spec: disableInitContainers: description: Disable the 'feast repo initialization' initContainer type: boolean + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -16532,6 +16621,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration applies + to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration matches + to. + type: string + type: object + type: array ui: description: Creates a UI server container properties: @@ -19204,6 +19327,13 @@ spec: disableInitContainers: description: Disable the 'feast repo initialization' initContainer type: boolean + nodeSelector: + additionalProperties: + type: string + description: |- + NodeSelector is a selector which must be true for the FeatureStore deployment + pods to fit on a node. + type: object offlineStore: description: OfflineStore configures the offline store service properties: @@ -21093,6 +21223,40 @@ spec: type: string type: object type: object + tolerations: + description: |- + Tolerations are applied to the FeatureStore deployment pods, allowing them to + be scheduled onto nodes with matching... + items: + description: |- + The pod this Toleration is attached to tolerates any taint that matches + the triple using the... + properties: + effect: + description: Effect indicates the taint effect to match. + Empty means match all taint effects. + type: string + key: + description: Key is the taint key that the toleration + applies to. Empty means match all taint keys. + type: string + operator: + description: |- + Operator represents a key's relationship to the value. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. + type: string + tolerationSeconds: + description: |- + TolerationSeconds represents the period of time the toleration (which must be + of effect NoExecute, otherwise this field... + format: int64 + type: integer + value: + description: Value is the taint value the toleration + matches to. + type: string + type: object + type: array ui: description: Creates a UI server container properties: diff --git a/infra/feast-operator/docs/api/markdown/ref.md b/infra/feast-operator/docs/api/markdown/ref.md index a303763fd7d..a69d6c98d90 100644 --- a/infra/feast-operator/docs/api/markdown/ref.md +++ b/infra/feast-operator/docs/api/markdown/ref.md @@ -292,6 +292,11 @@ Set to an empty array to disable auto-injection. | | `affinity` _[Affinity](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#affinity-v1-core)_ | Affinity defines the pod scheduling constraints for the FeatureStore deployment. When scaling is enabled and this is not set, the operator auto-injects a soft pod anti-affinity rule to prefer spreading pods across nodes. | +| `tolerations` _[Toleration](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#toleration-v1-core) array_ | Tolerations are applied to the FeatureStore deployment pods, allowing them to +be scheduled onto nodes with matching taints. | +| `nodeSelector` _object (keys:string, values:string)_ | NodeSelector is a selector which must be true for the FeatureStore deployment +pods to fit on a node. This selector must match a node's labels for the pod to +be scheduled on that node. | | `resourceClaims` _[PodResourceClaim](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#podresourceclaim-v1-core) array_ | ResourceClaims defines which ResourceClaims must be allocated and reserved before the Pod is allowed to start. The resources will be made available to those containers which consume them diff --git a/infra/feast-operator/internal/controller/services/services.go b/infra/feast-operator/internal/controller/services/services.go index 794754ce671..3ae2fa773f6 100644 --- a/infra/feast-operator/internal/controller/services/services.go +++ b/infra/feast-operator/internal/controller/services/services.go @@ -516,6 +516,7 @@ func (feast *FeastServices) setPod(podSpec *corev1.PodSpec) error { feast.mountEmptyDirVolumes(podSpec) feast.mountUserDefinedVolumes(podSpec) feast.applyNodeSelector(podSpec) + feast.applyTolerations(podSpec) feast.applyTopologySpread(podSpec) feast.applyAffinity(podSpec) feast.applyResourceClaims(podSpec) @@ -1142,8 +1143,18 @@ func (feast *FeastServices) getNodeSelectorForType(feastType FeastServiceType) * } func (feast *FeastServices) applyNodeSelector(podSpec *corev1.PodSpec) { - // Merge node selectors from all services + cr := feast.Handler.FeatureStore + services := cr.Status.Applied.Services + + // Start with the pod-level node selector configured on the FeatureStore + // services, then overlay per-service container config node selectors + // (per-service selectors win on key conflicts). mergedNodeSelector := make(map[string]string) + if services != nil && len(services.NodeSelector) > 0 { + for k, v := range services.NodeSelector { + mergedNodeSelector[k] = v + } + } // Check all service types for node selector configuration allServiceTypes := append(feastServerTypes, UIFeastType) @@ -1166,6 +1177,14 @@ func (feast *FeastServices) applyNodeSelector(podSpec *corev1.PodSpec) { podSpec.NodeSelector = finalNodeSelector } +func (feast *FeastServices) applyTolerations(podSpec *corev1.PodSpec) { + services := feast.Handler.FeatureStore.Status.Applied.Services + + if services != nil && services.Tolerations != nil { + podSpec.Tolerations = services.Tolerations + } +} + func (feast *FeastServices) applyTopologySpread(podSpec *corev1.PodSpec) { cr := feast.Handler.FeatureStore services := cr.Status.Applied.Services diff --git a/infra/feast-operator/internal/controller/services/services_test.go b/infra/feast-operator/internal/controller/services/services_test.go index da3590674f1..d0956148ae6 100644 --- a/infra/feast-operator/internal/controller/services/services_test.go +++ b/infra/feast-operator/internal/controller/services/services_test.go @@ -495,6 +495,56 @@ var _ = Describe("Registry Service", func() { Expect(deployment.Spec.Template.Spec.NodeSelector).To(Equal(expectedNodeSelector)) }) + It("should apply top-level NodeSelector to pod spec when configured", func() { + featureStore.Spec.Services.NodeSelector = map[string]string{ + kubernetesOsLabel: linuxOS, + nodeTypeLabel: computeNodeType, + } + Expect(k8sClient.Update(ctx, featureStore)).To(Succeed()) + Expect(feast.ApplyDefaults()).To(Succeed()) + applySpecToStatus(featureStore) + feast.refreshFeatureStore(ctx, typeNamespacedName) + + // Create deployment and verify NodeSelector is applied + deployment := feast.initFeastDeploy() + Expect(deployment).NotTo(BeNil()) + Expect(feast.setDeployment(deployment)).To(Succeed()) + + expectedNodeSelector := map[string]string{ + kubernetesOsLabel: linuxOS, + nodeTypeLabel: computeNodeType, + } + Expect(deployment.Spec.Template.Spec.NodeSelector).To(Equal(expectedNodeSelector)) + }) + + It("should let per-service NodeSelector override top-level NodeSelector on conflicting keys", func() { + featureStore.Spec.Services.NodeSelector = map[string]string{ + kubernetesOsLabel: linuxOS, + nodeTypeLabel: computeNodeType, + } + registryNodeSelector := map[string]string{ + nodeTypeLabel: "registry", + zoneLabel: "us-west-1a", + } + featureStore.Spec.Services.Registry.Local.Server.ContainerConfigs.OptionalCtrConfigs.NodeSelector = ®istryNodeSelector + Expect(k8sClient.Update(ctx, featureStore)).To(Succeed()) + Expect(feast.ApplyDefaults()).To(Succeed()) + applySpecToStatus(featureStore) + feast.refreshFeatureStore(ctx, typeNamespacedName) + + // Create deployment and verify merged NodeSelector is applied + deployment := feast.initFeastDeploy() + Expect(deployment).NotTo(BeNil()) + Expect(feast.setDeployment(deployment)).To(Succeed()) + + expectedNodeSelector := map[string]string{ + kubernetesOsLabel: linuxOS, + nodeTypeLabel: "registry", + zoneLabel: "us-west-1a", + } + Expect(deployment.Spec.Template.Spec.NodeSelector).To(Equal(expectedNodeSelector)) + }) + It("should enable metrics on the online service when configured", func() { featureStore.Spec.Services.OnlineStore = &feastdevv1.OnlineStore{ Server: &feastdevv1.ServerConfigs{Metrics: ptr.To(true)}, @@ -557,6 +607,45 @@ var _ = Describe("Registry Service", func() { }) }) + Describe("Tolerations Configuration", func() { + It("should apply Tolerations to pod spec when configured", func() { + tolerations := []corev1.Toleration{ + { + Key: "dedicated", + Operator: corev1.TolerationOpEqual, + Value: "feast", + Effect: corev1.TaintEffectNoSchedule, + }, + } + featureStore.Spec.Services.Tolerations = tolerations + Expect(k8sClient.Update(ctx, featureStore)).To(Succeed()) + Expect(feast.ApplyDefaults()).To(Succeed()) + applySpecToStatus(featureStore) + feast.refreshFeatureStore(ctx, typeNamespacedName) + + // Create deployment and verify Tolerations are applied + deployment := feast.initFeastDeploy() + Expect(deployment).NotTo(BeNil()) + Expect(feast.setDeployment(deployment)).To(Succeed()) + + Expect(deployment.Spec.Template.Spec.Tolerations).To(Equal(tolerations)) + }) + + It("should leave Tolerations empty when not configured", func() { + Expect(k8sClient.Update(ctx, featureStore)).To(Succeed()) + Expect(feast.ApplyDefaults()).To(Succeed()) + applySpecToStatus(featureStore) + feast.refreshFeatureStore(ctx, typeNamespacedName) + + // Create deployment and verify no Tolerations are applied + deployment := feast.initFeastDeploy() + Expect(deployment).NotTo(BeNil()) + Expect(feast.setDeployment(deployment)).To(Succeed()) + + Expect(deployment.Spec.Template.Spec.Tolerations).To(BeEmpty()) + }) + }) + Describe("WorkerConfigs Configuration", func() { It("should apply WorkerConfigs to the online store command", func() { // Set WorkerConfigs for online store diff --git a/infra/feast-operator/test/api/featurestore_v1alpha1_scheduling_test.go b/infra/feast-operator/test/api/featurestore_v1alpha1_scheduling_test.go new file mode 100644 index 00000000000..40a7f3c3aab --- /dev/null +++ b/infra/feast-operator/test/api/featurestore_v1alpha1_scheduling_test.go @@ -0,0 +1,62 @@ +/* +Copyright 2026 Feast Community. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +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. +*/ + +package api + +import ( + "context" + + feastdevv1alpha1 "github.com/feast-dev/feast/infra/feast-operator/api/v1alpha1" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +var _ = Describe("FeatureStore v1alpha1 scheduling configuration", func() { + It("accepts and preserves tolerations and nodeSelector", func() { + ctx := context.Background() + key := types.NamespacedName{Name: "v1alpha1-scheduling", Namespace: namespaceName} + expectedTolerations := []corev1.Toleration{{ + Key: "dedicated", + Operator: corev1.TolerationOpEqual, + Value: "feast", + Effect: corev1.TaintEffectNoSchedule, + }} + expectedNodeSelector := map[string]string{"kubernetes.io/os": "linux"} + featureStore := &feastdevv1alpha1.FeatureStore{ + ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace}, + Spec: feastdevv1alpha1.FeatureStoreSpec{ + FeastProject: "test_project", + Services: &feastdevv1alpha1.FeatureStoreServices{ + Tolerations: expectedTolerations, + NodeSelector: expectedNodeSelector, + }, + }, + } + + Expect(k8sClient.Create(ctx, featureStore)).To(Succeed()) + DeferCleanup(func() { + Expect(k8sClient.Delete(ctx, featureStore)).To(Succeed()) + }) + + actual := &feastdevv1alpha1.FeatureStore{} + Expect(k8sClient.Get(ctx, key, actual)).To(Succeed()) + Expect(actual.Spec.Services.Tolerations).To(Equal(expectedTolerations)) + Expect(actual.Spec.Services.NodeSelector).To(Equal(expectedNodeSelector)) + }) +}) From f0bc0700be7029166e6605e407de6d229cd1e93e Mon Sep 17 00:00:00 2001 From: Jitendra Yejare Date: Fri, 21 Aug 2026 21:18:11 +0530 Subject: [PATCH 2/2] fix: Feature freshness logic updated (#6765) * fix: feature freshness logic updated Signed-off-by: Jitendra Yejare <11752425+jyejare@users.noreply.github.com> * fix: Avoid redundant _get_max_timestamp query in _compute_for_feature_view Move the _get_max_timestamp call to compute_metrics and pass the result into _compute_for_feature_view, eliminating a duplicate data-source query per feature view. Signed-off-by: Jitendra Yejare <11752425+jyejare@users.noreply.github.com> Co-authored-by: Cursor --------- Signed-off-by: Jitendra Yejare <11752425+jyejare@users.noreply.github.com> Co-authored-by: Cursor --- docs/how-to-guides/feature-monitoring.md | 3 + .../feast/infra/offline_stores/bigquery.py | 10 ++- .../contrib/oracle_offline_store/oracle.py | 16 ++++ .../postgres_offline_store/postgres.py | 12 +++ .../contrib/spark_offline_store/spark.py | 15 ++++ sdk/python/feast/infra/offline_stores/dask.py | 3 +- .../feast/infra/offline_stores/duckdb.py | 3 +- .../feast/infra/offline_stores/redshift.py | 6 ++ .../feast/infra/offline_stores/snowflake.py | 8 ++ .../feast/monitoring/monitoring_service.py | 67 +++++++++++++++ .../feast/monitoring/monitoring_utils.py | 21 ++++- .../monitoring/test_monitoring_integration.py | 10 +++ .../unit/monitoring/test_feature_freshness.py | 85 +++++++++++++++++++ .../pages/monitoring/FeatureMetricsTable.tsx | 38 +++++---- ui/src/queries/useMonitoringApi.ts | 4 + 15 files changed, 281 insertions(+), 20 deletions(-) create mode 100644 sdk/python/tests/unit/monitoring/test_feature_freshness.py diff --git a/docs/how-to-guides/feature-monitoring.md b/docs/how-to-guides/feature-monitoring.md index aca36167323..fc8c693ed30 100644 --- a/docs/how-to-guides/feature-monitoring.md +++ b/docs/how-to-guides/feature-monitoring.md @@ -219,6 +219,7 @@ GET /monitoring/metrics/features?project=my_project&feature_view_name=driver_sta "feature_name": "conv_rate", "feature_type": "numeric", "metric_date": "2025-03-26", + "max_event_timestamp": "2025-03-27T14:30:00+00:00", "granularity": "daily", "data_source_type": "batch", "row_count": 15000, @@ -242,6 +243,8 @@ GET /monitoring/metrics/features?project=my_project&feature_view_name=driver_sta ] ``` +The UI **Freshness** column uses `max_event_timestamp` — `MAX(event_timestamp)` from the source — not `metric_date` (the DQM window start). Age is `now − max_event_timestamp`. + ### Per-feature-view aggregates ``` diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index 269572dd9da..afe3e6f97cc 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -693,7 +693,7 @@ def _bq_scalar_param_type(column: str) -> str: return "BOOL" if column == "metric_date": return "DATE" - if column == "computed_at": + if column in ("computed_at", "max_event_timestamp"): return "TIMESTAMP" if column in { "row_count", @@ -889,6 +889,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None: granularity STRING NOT NULL, data_source_type STRING NOT NULL, computed_at TIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP, is_baseline BOOL NOT NULL, feature_type STRING NOT NULL, row_count INT64, @@ -915,6 +916,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None: granularity STRING NOT NULL, data_source_type STRING NOT NULL, computed_at TIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP, is_baseline BOOL NOT NULL, total_row_count INT64, total_features INT64, @@ -932,6 +934,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None: granularity STRING NOT NULL, data_source_type STRING NOT NULL, computed_at TIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP, is_baseline BOOL NOT NULL, total_feature_views INT64, total_features INT64, @@ -958,6 +961,11 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None: """ for ddl in (feature_ddl, view_ddl, service_ddl, job_ddl): client.query(ddl).result() + for tbl in (MON_TABLE_FEATURE, MON_TABLE_FEATURE_VIEW, MON_TABLE_FEATURE_SERVICE): + client.query( + f"ALTER TABLE `{proj}.{ds}.{tbl}` " + "ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMP" + ).result() def _bq_get_monitoring_max_timestamp( diff --git a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py index 0aa657c69c9..6fb723e0786 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py +++ b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py @@ -715,6 +715,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL, data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL, computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP WITH TIME ZONE, is_baseline NUMBER(1) DEFAULT 0 NOT NULL, feature_type VARCHAR2(50) NOT NULL, row_count NUMBER, @@ -746,6 +747,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL, data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL, computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP WITH TIME ZONE, is_baseline NUMBER(1) DEFAULT 0 NOT NULL, total_row_count NUMBER, total_features NUMBER, @@ -768,6 +770,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL, data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL, computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP WITH TIME ZONE, is_baseline NUMBER(1) DEFAULT 0 NOT NULL, total_feature_views NUMBER, total_features NUMBER, @@ -779,6 +782,19 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: """, ) + _oracle_try_execute_ddl( + con, + f"ALTER TABLE {MON_TABLE_FEATURE} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)", + ) + _oracle_try_execute_ddl( + con, + f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)", + ) + _oracle_try_execute_ddl( + con, + f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)", + ) + _oracle_try_execute_ddl( con, f""" diff --git a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py index 63363715cd6..79fba09388f 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py +++ b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py @@ -429,6 +429,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: granularity VARCHAR(20) NOT NULL DEFAULT 'daily', data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch', computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + max_event_timestamp TIMESTAMPTZ, is_baseline BOOLEAN NOT NULL DEFAULT FALSE, feature_type VARCHAR(50) NOT NULL, row_count BIGINT, @@ -468,6 +469,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: granularity VARCHAR(20) NOT NULL DEFAULT 'daily', data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch', computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + max_event_timestamp TIMESTAMPTZ, is_baseline BOOLEAN NOT NULL DEFAULT FALSE, total_row_count BIGINT, total_features INTEGER, @@ -487,6 +489,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: granularity VARCHAR(20) NOT NULL DEFAULT 'daily', data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch', computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + max_event_timestamp TIMESTAMPTZ, is_baseline BOOLEAN NOT NULL DEFAULT FALSE, total_feature_views INTEGER, total_features INTEGER, @@ -497,6 +500,15 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: ); """) + cur.execute(f""" + ALTER TABLE {MON_TABLE_FEATURE} + ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ; + ALTER TABLE {MON_TABLE_FEATURE_VIEW} + ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ; + ALTER TABLE {MON_TABLE_FEATURE_SERVICE} + ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ; + """) + cur.execute(f""" CREATE TABLE IF NOT EXISTS {MON_TABLE_JOB} ( job_id VARCHAR(36) PRIMARY KEY, diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py index 7e0a03e69bb..b891eac6a80 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py +++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py @@ -574,6 +574,18 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: ) for stmt in _SPARK_MONITORING_DDL_STATEMENTS: spark_session.sql(stmt) + from pyspark.sql.utils import AnalysisException + + for stmt in ( + f"ALTER TABLE {MON_TABLE_FEATURE} ADD COLUMNS (max_event_timestamp TIMESTAMP)", + f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD COLUMNS (max_event_timestamp TIMESTAMP)", + f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD COLUMNS (max_event_timestamp TIMESTAMP)", + ): + try: + spark_session.sql(stmt) + except AnalysisException: + # Column already exists on newly created tables. + pass @staticmethod def save_monitoring_metrics( @@ -678,6 +690,7 @@ def clear_monitoring_baseline( granularity STRING NOT NULL, data_source_type STRING NOT NULL, computed_at TIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP, is_baseline BOOLEAN NOT NULL, feature_type STRING NOT NULL, row_count BIGINT, @@ -703,6 +716,7 @@ def clear_monitoring_baseline( granularity STRING NOT NULL, data_source_type STRING NOT NULL, computed_at TIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP, is_baseline BOOLEAN NOT NULL, total_row_count BIGINT, total_features INT, @@ -719,6 +733,7 @@ def clear_monitoring_baseline( granularity STRING NOT NULL, data_source_type STRING NOT NULL, computed_at TIMESTAMP NOT NULL, + max_event_timestamp TIMESTAMP, is_baseline BOOLEAN NOT NULL, total_feature_views INT, total_features INT, diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index f9d22250c7d..cd930f7ab5c 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -40,6 +40,7 @@ from feast.monitoring.monitoring_utils import ( MONITORING_DIR, MONITORING_PARQUET_FILES, + MONITORING_TIMESTAMP_FIELDS, monitoring_parquet_meta, normalize_monitoring_row, opt_float, @@ -968,7 +969,7 @@ def _dask_parquet_query( for _, row in df.iterrows(): record = {c: row.get(c) for c in columns} normalize_monitoring_row(record) - for key in ("metric_date", "computed_at"): + for key in MONITORING_TIMESTAMP_FIELDS: val = record.get(key) if ( val is not None diff --git a/sdk/python/feast/infra/offline_stores/duckdb.py b/sdk/python/feast/infra/offline_stores/duckdb.py index 7439f8bb1ea..1ff09d9f3cf 100644 --- a/sdk/python/feast/infra/offline_stores/duckdb.py +++ b/sdk/python/feast/infra/offline_stores/duckdb.py @@ -39,6 +39,7 @@ from feast.monitoring.monitoring_utils import ( MONITORING_DIR, MONITORING_PARQUET_FILES, + MONITORING_TIMESTAMP_FIELDS, empty_categorical_metric, empty_numeric_metric, monitoring_parquet_meta, @@ -487,7 +488,7 @@ def _duckdb_parquet_query( for _, row in df.iterrows(): record = {c: row.get(c) for c in columns} normalize_monitoring_row(record) - for key in ("metric_date", "computed_at"): + for key in MONITORING_TIMESTAMP_FIELDS: val = record.get(key) if ( val is not None diff --git a/sdk/python/feast/infra/offline_stores/redshift.py b/sdk/python/feast/infra/offline_stores/redshift.py index 5562c233806..76effbf2aa6 100644 --- a/sdk/python/feast/infra/offline_stores/redshift.py +++ b/sdk/python/feast/infra/offline_stores/redshift.py @@ -555,6 +555,7 @@ def clear_monitoring_baseline( granularity VARCHAR(20) NOT NULL DEFAULT 'daily', data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch', computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + max_event_timestamp TIMESTAMPTZ, is_baseline BOOLEAN NOT NULL DEFAULT FALSE, feature_type VARCHAR(50) NOT NULL, row_count BIGINT, @@ -582,6 +583,7 @@ def clear_monitoring_baseline( granularity VARCHAR(20) NOT NULL DEFAULT 'daily', data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch', computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + max_event_timestamp TIMESTAMPTZ, is_baseline BOOLEAN NOT NULL DEFAULT FALSE, total_row_count BIGINT, total_features INTEGER, @@ -600,6 +602,7 @@ def clear_monitoring_baseline( granularity VARCHAR(20) NOT NULL DEFAULT 'daily', data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch', computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + max_event_timestamp TIMESTAMPTZ, is_baseline BOOLEAN NOT NULL DEFAULT FALSE, total_feature_views INTEGER, total_features INTEGER, @@ -625,6 +628,9 @@ def clear_monitoring_baseline( PRIMARY KEY (job_id) ); """, + f"ALTER TABLE {MON_TABLE_FEATURE} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ", + f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ", + f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ", ] diff --git a/sdk/python/feast/infra/offline_stores/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index 65b0b42b617..81430045f9e 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -564,6 +564,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: "granularity" VARCHAR(20) NOT NULL DEFAULT 'daily', "data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch', "computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(), + "max_event_timestamp" TIMESTAMP_TZ, "is_baseline" BOOLEAN NOT NULL DEFAULT FALSE, "feature_type" VARCHAR(50) NOT NULL, "row_count" BIGINT, @@ -591,6 +592,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: "granularity" VARCHAR(20) NOT NULL DEFAULT 'daily', "data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch', "computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(), + "max_event_timestamp" TIMESTAMP_TZ, "is_baseline" BOOLEAN NOT NULL DEFAULT FALSE, "total_row_count" BIGINT, "total_features" INTEGER, @@ -609,6 +611,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: "granularity" VARCHAR(20) NOT NULL DEFAULT 'daily', "data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch', "computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(), + "max_event_timestamp" TIMESTAMP_TZ, "is_baseline" BOOLEAN NOT NULL DEFAULT FALSE, "total_feature_views" INTEGER, "total_features" INTEGER, @@ -642,6 +645,11 @@ def ensure_monitoring_tables(config: RepoConfig) -> None: execute_snowflake_statement(conn, ddl_view) execute_snowflake_statement(conn, ddl_service) execute_snowflake_statement(conn, ddl_job) + for fq in (fq_feature, fq_view, fq_service): + execute_snowflake_statement( + conn, + f'ALTER TABLE {fq} ADD COLUMN IF NOT EXISTS "max_event_timestamp" TIMESTAMP_TZ', + ) @staticmethod def save_monitoring_metrics( diff --git a/sdk/python/feast/monitoring/monitoring_service.py b/sdk/python/feast/monitoring/monitoring_service.py index df6c16175f6..aa0711cb990 100644 --- a/sdk/python/feast/monitoring/monitoring_service.py +++ b/sdk/python/feast/monitoring/monitoring_service.py @@ -29,6 +29,37 @@ "quarterly": timedelta(days=90), } + +def _as_utc_datetime(val: Any) -> Optional[datetime]: + """Parse a timestamp-like value to a timezone-aware UTC datetime.""" + if val is None: + return None + if isinstance(val, datetime): + return val if val.tzinfo else val.replace(tzinfo=timezone.utc) + if isinstance(val, date): + return datetime.combine(val, datetime.min.time(), tzinfo=timezone.utc) + if isinstance(val, str): + parsed = datetime.fromisoformat(val.replace("Z", "+00:00")) + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + return None + + +def _newest_event_in_window( + max_ts: Optional[datetime], + start_dt: datetime, + end_dt: datetime, +) -> Optional[datetime]: + """Newest source event that falls inside ``[start_dt, end_dt]``. + + ``max_ts`` is MAX(event_timestamp) over the whole source. If that + timestamp is after the window, the window end is used as an upper bound. + """ + aware_max = _as_utc_datetime(max_ts) + if aware_max is None or aware_max < start_dt: + return None + return min(aware_max, end_dt) + + _FLOAT_FIELDS = frozenset( { "null_rate", @@ -146,6 +177,7 @@ def auto_compute( granularity="baseline", set_baseline=True, now=now, + max_event_timestamp=max_ts, ) baseline_features += len(bl_metrics) @@ -165,6 +197,7 @@ def auto_compute( granularity=granularity, set_baseline=False, now=now, + max_event_timestamp=max_ts, ) self._compute_feature_service_metrics( project=project, @@ -251,6 +284,11 @@ def compute_log_metrics( granularity=granularity, set_baseline=set_baseline, now=now, + max_event_timestamp=_newest_event_in_window( + self._get_max_timestamp_for_source(data_source, ts_field), + start_dt, + end_dt, + ), ) duration_ms = int((time.time() - start_time) * 1000) @@ -324,6 +362,7 @@ def auto_compute_log_metrics( granularity=gran, set_baseline=False, now=now, + max_event_timestamp=max_ts, ) total_features += len(metrics_list) granularities_computed.add(gran) @@ -402,6 +441,7 @@ def compute_baseline( granularity="baseline", set_baseline=True, now=now, + max_event_timestamp=self._get_max_timestamp(fv), ) total_features += len(metrics_list) @@ -453,6 +493,7 @@ def compute_metrics( for fv in feature_views: try: + max_ts = self._get_max_timestamp(fv) fv_metrics = self._compute_for_feature_view( project=project, feature_view=fv, @@ -461,6 +502,7 @@ def compute_metrics( end_dt=end_dt, granularity=granularity, set_baseline=set_baseline, + max_event_timestamp=max_ts, ) total_features += fv_metrics["feature_count"] total_views += 1 @@ -834,6 +876,7 @@ def _save_computed_metrics( granularity: str, set_baseline: bool, now: datetime, + max_event_timestamp: Optional[datetime] = None, ) -> None: if not metrics_list: return @@ -855,6 +898,7 @@ def _save_computed_metrics( m["granularity"] = granularity m["data_source_type"] = "batch" m["computed_at"] = now + m["max_event_timestamp"] = max_event_timestamp m["is_baseline"] = set_baseline offline_store.save_monitoring_metrics(config, "feature", metrics_list) @@ -866,6 +910,7 @@ def _save_computed_metrics( "granularity": granularity, "data_source_type": "batch", "computed_at": now, + "max_event_timestamp": max_event_timestamp, "is_baseline": set_baseline, **build_view_aggregate(metrics_list), } @@ -925,6 +970,7 @@ def _compute_for_feature_view( end_dt: datetime, granularity: str, set_baseline: bool, + max_event_timestamp: Optional[datetime] = None, ) -> Dict[str, Any]: feature_fields = self._classify_fields( feature_view, feature_names=feature_names @@ -950,6 +996,11 @@ def _compute_for_feature_view( granularity=granularity, set_baseline=set_baseline, now=now, + max_event_timestamp=_newest_event_in_window( + max_event_timestamp, + start_dt, + end_dt, + ), ) return {"feature_count": len(metrics_list), "dates": {metric_date}} @@ -1119,6 +1170,7 @@ def _save_log_metrics( granularity: str, set_baseline: bool, now: datetime, + max_event_timestamp: Optional[datetime] = None, ) -> None: """Save log-sourced metrics tagged with data_source_type='log'. @@ -1145,6 +1197,7 @@ def _save_log_metrics( m["granularity"] = granularity m["data_source_type"] = "log" m["computed_at"] = now + m["max_event_timestamp"] = max_event_timestamp m["is_baseline"] = set_baseline offline_store.save_monitoring_metrics(config, "feature", metrics_list) @@ -1162,6 +1215,7 @@ def _save_log_metrics( "granularity": granularity, "data_source_type": "log", "computed_at": now, + "max_event_timestamp": max_event_timestamp, "is_baseline": set_baseline, **build_view_aggregate(vmetrics), } @@ -1178,6 +1232,7 @@ def _save_log_metrics( "granularity": granularity, "data_source_type": "log", "computed_at": now, + "max_event_timestamp": max_event_timestamp, "is_baseline": set_baseline, "total_feature_views": len(by_view), "total_features": svc_agg["total_features"], @@ -1252,6 +1307,15 @@ def _compute_feature_service_metrics( if m.get("avg_null_rate") is not None ] + newest_events = [ + ts + for ts in ( + _as_utc_datetime(m.get("max_event_timestamp")) + for m in relevant + ) + if ts is not None + ] + service_metric = { "project_id": project, "feature_service_name": fs.name, @@ -1261,6 +1325,9 @@ def _compute_feature_service_metrics( "granularity": granularity, "data_source_type": "batch", "computed_at": now, + "max_event_timestamp": max(newest_events) + if newest_events + else None, "is_baseline": set_baseline, "total_feature_views": len(relevant), "total_features": sum( diff --git a/sdk/python/feast/monitoring/monitoring_utils.py b/sdk/python/feast/monitoring/monitoring_utils.py index 0450e008a05..2b4bc2d8b52 100644 --- a/sdk/python/feast/monitoring/monitoring_utils.py +++ b/sdk/python/feast/monitoring/monitoring_utils.py @@ -32,6 +32,13 @@ # Column definitions — (ordered, used by INSERT / SELECT / Parquet) # ------------------------------------------------------------------ # +# Datetime fields serialized to ISO-8601 on read. +MONITORING_TIMESTAMP_FIELDS: Tuple[str, ...] = ( + "metric_date", + "computed_at", + "max_event_timestamp", +) + FEATURE_METRICS_COLUMNS: List[str] = [ "project_id", "feature_view_name", @@ -40,6 +47,7 @@ "granularity", "data_source_type", "computed_at", + "max_event_timestamp", "is_baseline", "feature_type", "row_count", @@ -73,6 +81,7 @@ "granularity", "data_source_type", "computed_at", + "max_event_timestamp", "is_baseline", "total_row_count", "total_features", @@ -96,6 +105,7 @@ "granularity", "data_source_type", "computed_at", + "max_event_timestamp", "is_baseline", "total_feature_views", "total_features", @@ -238,7 +248,8 @@ def normalize_monitoring_row(record: Dict[str, Any]) -> Dict[str, Any]: - Replaces float NaN / Inf with None (not JSON-serializable). - Parses ``histogram`` from JSON string if needed. - - Converts ``metric_date`` / ``computed_at`` to ISO strings. + - Converts ``metric_date`` / ``computed_at`` / ``max_event_timestamp`` + to ISO strings. - Normalizes ``is_baseline`` to Python bool. """ import math @@ -254,8 +265,14 @@ def normalize_monitoring_row(record: Dict[str, Any]) -> Dict[str, Any]: except (json.JSONDecodeError, TypeError): pass - for key in ("metric_date", "computed_at"): + for key in MONITORING_TIMESTAMP_FIELDS: val = record.get(key) + if val is None: + continue + # pandas NaT / NaN leak through parquet reads as non-datetime sentinels. + if val is not val or str(val) == "NaT": + record[key] = None + continue if isinstance(val, (date, datetime)): record[key] = val.isoformat() diff --git a/sdk/python/tests/integration/monitoring/test_monitoring_integration.py b/sdk/python/tests/integration/monitoring/test_monitoring_integration.py index 59e045bf0b5..41a48530917 100644 --- a/sdk/python/tests/integration/monitoring/test_monitoring_integration.py +++ b/sdk/python/tests/integration/monitoring/test_monitoring_integration.py @@ -780,6 +780,16 @@ def test_auto_compute_uses_pushdown_for_max_timestamp(self): provider.offline_store.compute_monitoring_metrics.assert_called() provider.offline_store.pull_all_from_table_or_query.assert_not_called() + newest = datetime(2025, 3, 27, tzinfo=timezone.utc) + feature_saves = [ + call + for call in provider.offline_store.save_monitoring_metrics.call_args_list + if call.args[1] == "feature" + ] + assert feature_saves + saved = feature_saves[0].args[2] + assert all(row["max_event_timestamp"] == newest for row in saved) + # ------------------------------------------------------------------ # # Test: Native storage dispatch diff --git a/sdk/python/tests/unit/monitoring/test_feature_freshness.py b/sdk/python/tests/unit/monitoring/test_feature_freshness.py new file mode 100644 index 00000000000..1b3bd9d24d7 --- /dev/null +++ b/sdk/python/tests/unit/monitoring/test_feature_freshness.py @@ -0,0 +1,85 @@ +from datetime import datetime, timezone +from unittest.mock import MagicMock + +from feast.monitoring.monitoring_service import ( + MonitoringService, + _newest_event_in_window, +) +from feast.types import PrimitiveFeastType + + +class TestNewestEventInWindow: + def test_uses_source_max_when_inside_window(self): + max_ts = datetime(2025, 3, 26, 14, 30, tzinfo=timezone.utc) + start = datetime(2025, 3, 25, 14, 30, tzinfo=timezone.utc) + end = datetime(2025, 3, 26, 14, 30, tzinfo=timezone.utc) + assert _newest_event_in_window(max_ts, start, end) == max_ts + + def test_clamps_to_window_end_when_source_is_newer(self): + max_ts = datetime(2025, 4, 1, tzinfo=timezone.utc) + start = datetime(2025, 1, 1, tzinfo=timezone.utc) + end = datetime(2025, 1, 15, tzinfo=timezone.utc) + assert _newest_event_in_window(max_ts, start, end) == end + + def test_returns_none_when_source_is_before_window(self): + max_ts = datetime(2024, 12, 1, tzinfo=timezone.utc) + start = datetime(2025, 1, 1, tzinfo=timezone.utc) + end = datetime(2025, 1, 15, tzinfo=timezone.utc) + assert _newest_event_in_window(max_ts, start, end) is None + + +def test_auto_compute_persists_newest_event_timestamp(): + field = MagicMock() + field.name = "conv_rate" + field.dtype = PrimitiveFeastType.FLOAT64 + fv = MagicMock() + fv.name = "driver_stats" + fv.features = [field] + fv.entities = [] + fv.batch_source.timestamp_field = "event_timestamp" + fv.batch_source.created_timestamp_column = "" + + store = MagicMock() + store.config.project = "test_project" + store.registry.list_feature_views.return_value = [fv] + store.registry.list_entities.return_value = [] + store.registry.list_feature_services.return_value = [] + store.registry.get_feature_view.return_value = fv + + newest = datetime(2025, 3, 27, 14, 30, tzinfo=timezone.utc) + provider = store._get_provider.return_value + provider.offline_store.get_monitoring_max_timestamp.side_effect = None + provider.offline_store.get_monitoring_max_timestamp.return_value = newest + provider.offline_store.compute_monitoring_metrics.side_effect = None + provider.offline_store.compute_monitoring_metrics.return_value = [ + { + "feature_name": "conv_rate", + "feature_type": "numeric", + "row_count": 5, + "null_count": 0, + "null_rate": 0.0, + "mean": 0.5, + "stddev": 0.2, + "min_val": 0.1, + "max_val": 0.9, + "p50": 0.5, + "p75": 0.7, + "p90": 0.9, + "p95": 0.9, + "p99": 0.9, + "histogram": None, + }, + ] + provider.offline_store.query_monitoring_metrics.return_value = [] + + result = MonitoringService(store).auto_compute(project="test_project") + assert result["status"] == "completed" + + feature_saves = [ + call + for call in provider.offline_store.save_monitoring_metrics.call_args_list + if call.args[1] == "feature" + ] + assert feature_saves + saved = feature_saves[0].args[2] + assert all(row["max_event_timestamp"] == newest for row in saved) diff --git a/ui/src/pages/monitoring/FeatureMetricsTable.tsx b/ui/src/pages/monitoring/FeatureMetricsTable.tsx index 5b998c98aee..c3f775757ee 100644 --- a/ui/src/pages/monitoring/FeatureMetricsTable.tsx +++ b/ui/src/pages/monitoring/FeatureMetricsTable.tsx @@ -39,9 +39,10 @@ const formatNum = (val: number | null, decimals = 2): string => { return val.toFixed(decimals); }; -const formatFreshness = (computedAt: string | null): string => { - if (!computedAt) return "—"; - const diff = Date.now() - new Date(computedAt).getTime(); +const formatFreshness = (timestamp: string | null): string => { + if (!timestamp) return "—"; + const diff = Date.now() - new Date(timestamp).getTime(); + if (Number.isNaN(diff)) return "—"; const mins = Math.floor(diff / 60_000); if (mins < 1) return "just now"; if (mins < 60) return `${mins}m ago`; @@ -52,14 +53,18 @@ const formatFreshness = (computedAt: string | null): string => { return `${Math.floor(days / 30)}mo ago`; }; -const freshnessColor = (computedAt: string | null): string => { - if (!computedAt) return "subdued"; - const hrs = (Date.now() - new Date(computedAt).getTime()) / 3_600_000; +const freshnessColor = (timestamp: string | null): string => { + if (!timestamp) return "subdued"; + const hrs = (Date.now() - new Date(timestamp).getTime()) / 3_600_000; + if (Number.isNaN(hrs)) return "subdued"; if (hrs < 24) return "success"; if (hrs < 72) return "warning"; return "danger"; }; +const freshnessTimestamp = (metric: FeatureMetric): string | null => + metric.max_event_timestamp || metric.metric_date || null; + const MiniHistogram = ({ metric }: { metric: FeatureMetric }) => { if (!metric.histogram) return —; @@ -257,7 +262,7 @@ const FeatureMetricsTable = ({ { title: "Freshness", description: - "Recency of the underlying data. Green (< 24h old), Yellow (24–72h), Red (> 72h). Hover for the data date.", + "Age of the newest source event (MAX of the event timestamp). Green (< 24h old), Yellow (24–72h), Red (> 72h). Hover for the exact event time.", }, { title: "Source", @@ -346,17 +351,20 @@ const FeatureMetricsTable = ({ render: (val: number | null) => formatNum(val), }, { - field: "metric_date", + field: "max_event_timestamp", name: "Freshness", sortable: true, width: "110px", - render: (val: string) => ( - - - {formatFreshness(val)} - - - ), + render: (_val: string | null, item: FeatureMetric) => { + const ts = freshnessTimestamp(item); + return ( + + + {formatFreshness(ts)} + + + ); + }, }, { field: "data_source_type", diff --git a/ui/src/queries/useMonitoringApi.ts b/ui/src/queries/useMonitoringApi.ts index 73a9b16e3fd..a390b50541b 100644 --- a/ui/src/queries/useMonitoringApi.ts +++ b/ui/src/queries/useMonitoringApi.ts @@ -12,6 +12,7 @@ interface FeatureMetric { granularity: string; data_source_type: string; computed_at: string; + max_event_timestamp: string | null; is_baseline: boolean; feature_type: string; row_count: number; @@ -48,6 +49,7 @@ interface FeatureViewMetric { granularity: string; data_source_type: string; computed_at: string; + max_event_timestamp: string | null; is_baseline: boolean; total_row_count: number; total_features: number; @@ -63,6 +65,7 @@ interface FeatureServiceMetric { granularity: string; data_source_type: string; computed_at: string; + max_event_timestamp: string | null; is_baseline: boolean; total_feature_views: number; total_features: number; @@ -188,6 +191,7 @@ const aggregateToFeatureViewMetrics = ( granularity: feats[0].granularity, data_source_type: feats[0].data_source_type, computed_at: feats[0].computed_at, + max_event_timestamp: feats[0].max_event_timestamp, is_baseline: feats[0].is_baseline, total_row_count: maxRowCount, total_features: feats.length,