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/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/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)) + }) +}) diff --git a/infra/website/docs/blog/aerospike-feast-benchmark-harness-results.md b/infra/website/docs/blog/aerospike-feast-benchmark-harness-results.md new file mode 100644 index 00000000000..58101c36bfe --- /dev/null +++ b/infra/website/docs/blog/aerospike-feast-benchmark-harness-results.md @@ -0,0 +1,149 @@ +--- +title: "Fast like a cache, priced like storage: Benchmarking Aerospike on Feast" +description: "We ran Feast's benchmark harness against Aerospike and compared it with the published Redis and DynamoDB results, sweeping entities per request, features per request, and request rate." +date: 2026-08-20 +authors: ["Valentyn Kahamlyk", "Francisco Javier Arceo"] +--- + +
+ P99 latency with entity growth: Aerospike, Redis, and DynamoDB +
+ +We recently [launched an integration between Aerospike and Feast](/blog/aerospike-feast-now-available), giving teams a straightforward way to use Aerospike as the online store behind Feast's real-time feature serving. As part of that work, we wanted to understand how Aerospike performs under the workloads that matter for Feast users. + +We benchmarked Aerospike as a Feast online store using [Feast’s benchmark harness](https://github.com/feast-dev/feast-benchmarks) and compared the results with the Redis and DynamoDB results published in the same repository. The tests varied the number of entities per request, the number of features per request, and the request rate. + +## Benchmark methodology + +We used Feast's benchmark harness end-to-end, from the load generator through the Python feature server to the online store. + +We tested three dimensions of the workload: + +1. **Entities per request.** A single-entity request represents a scoring call for one user or transaction. A request containing many entities represents batch scoring or a model evaluating many candidates at once. +2. **Features per request.** More features represent a richer model and a larger amount of data that must be retrieved for each prediction. +3. **Request rate.** Increasing the request rate tests how each online store behaves as concurrency and load increase. + +Sweeping these axes separately allows us to observe how a store behaves under specific workload shapes rather than relying on a single latency number. + +### A note on methodology and reproducibility + +Teams can use the publicly available [harness](https://github.com/feast-dev/feast-benchmarks) to run the tests on their own hardware and workloads. The repository publishes results for several online stores. We reviewed all of them and compared Aerospike against [Redis](https://aerospike.com/compare/redis-vs-aerospike/) and [DynamoDB](https://aerospike.com/compare/dynamodb-vs-aerospike/), setting Datastore aside because its data is several years old and the service has since been rebuilt as Firestore in Datastore mode. + +The Aerospike tests used a c2-standard-16 GCP VM. Redis and Aerospike were co-located with the feature server, while DynamoDB was accessed as a managed service over a same-region network connection. We reproduced the community harness as published rather than running a controlled benchmark of our own, so running the load generator, feature server, and store together on one VM reflects that harness convention rather than a production topology. + +Because competitor results were collected on different hardware and at different times, we focus on workload behavior rather than precise latency multiples. These results should be read as workload comparisons rather than universal performance claims. + +## Measuring latency across request sizes + +The first set of tests increases the number of entities requested at once. + +A single-entity request is the common case, a fraud check or a personalization call scoring one user. Larger requests show up in recommendation and bidding systems that score many candidates at once to pick among them, so a single prediction can require tens or hundreds of entities in one read. + +Aerospike tracks the in-memory store closely on these ordinary reads. On single-entity and modest-batch requests, Aerospike and Redis sit together in the low tens of milliseconds, with the lead moving between runs within normal measurement variance. + +DynamoDB is several times higher in these tests. + +![P99 latency with entity growth: Aerospike, Redis, and DynamoDB](/images/blog/aerospike-benchmark-p99-latency-entity-growth.png) + +**Chart 1: Aerospike P99 latency stays relatively low as more entities are added** +*Note: Features \= 50, RPS \= 10 for all runs* + +This matters because Aerospike is not keeping the entire dataset in RAM. Its [Hybrid Memory Architecture (HMA)](https://aerospike.com/blog/hybrid-memory-architecture-optimization/) keeps primary indexes in memory while storing feature data on SSD. + +A read that an in-memory store handles quickly, Aerospike can handle in a similar latency range without requiring an all-in-memory deployment. + +## Increasing the number of features + +The second sweep increases the number of features retrieved for each entity. + +As the feature count increases, each request contains more data and places more work on the online store. Feature counts climb as teams add signals to a model: a mature fraud model may pull a wide set of velocity, history, and device features about a single entity, which is why the widest requests are realistic and not a stress-test artifact. + +![P99 latency with feature growth: Aerospike, Redis, and DynamoDB](/images/blog/aerospike-benchmark-p99-latency-feature-growth.png) + +**Chart 2: Aerospike P99 latency remains low as more features are added** +*Note: Entities \= 1, RPS \= 10 for all runs* + +As the request gets wider, Aerospike stays in the same low range as the in-memory store, and both stay well below DynamoDB. A production feature store needs to maintain predictable serving behavior as feature sets become larger. + +## Holding performance under load + +The third sweep increases the request rate, which is driven by traffic, not by the model. An ad-bidding or fraud-scoring service, at peak, fields thousands of decisions per second, and each decision can be one of these multi-entity reads, so entity-batch size and request rate rise together rather than independently. + +The harness includes running 100 entities and 50 features per request at increasing requests per second (RPS). + +* Aerospike maintains 100 percent success across the full range tested. +* Redis holds to about 60 requests per second before success rates begin to fall. At 80 requests per second, only about a quarter of requests succeed, and at 90 requests per second success approaches zero. +* DynamoDB is degraded from the first step, with about 72 percent success even at 10 requests per second. In this case, the limiting behavior is throttling rather than latency. + +The failures under load are different. Throttling and timeouts represent the behavior of the stores under the tested workloads and do not depend on the same topology effect. + +![Success rate with growing requests per second: Aerospike, Redis, and DynamoDB](/images/blog/aerospike-benchmark-success-rate-rps.png) + +**Chart 3: Aerospike successfully completes all runs up to the max 100 RPS** +*Note: Entities \= 100, Features \= 50 for all runs* + +Aerospike keeps serving successfully as both request size and request rate increase, while the other stores stop completing requests. + +## Behavior at the highest scale + +We also tested a request shape of 100 entities by 250 features. At this workload, Aerospike is the only store that completes any requests at all, and only at the lowest request rates. + +That result marks the practical ceiling of the complete serving stack under this test configuration rather than the maximum capability of any individual database. The feature server, network, client, and load generator all contribute to the result. + +![Success rate with growing requests per second at 100 entities by 250 features](/images/blog/aerospike-benchmark-success-rate-rps-wide-requests.png) + +**Chart 4: Redis and DynamoDB are unable to complete these runs. Aerospike succeeds up to 30 RPS** +*Note: Entities \= 100, Features \= 250 for all runs* + +This test is useful for a different reason than the smaller requests. It shows what happens when the amount of data required for each request becomes large enough that the serving stack itself becomes the limiting factor. + +## What the results mean + +Taken together, the tests show a consistent pattern. Aerospike delivers near in-memory latency on smaller feature requests while maintaining successful serving as request sizes and rates increase. + +That combination matters for feature stores because the workload can change in both dimensions over time. A model may start with a relatively small feature set and later add more features. An application may begin with modest traffic and eventually serve millions of predictions. + +Aerospike’s HMA addresses these requirements: indexes remain in memory for fast lookups, while feature data can reside on SSD. The performance results address one side of the tradeoff. The other is infrastructure cost: how much memory is required to deliver that performance as the feature set grows? + +## The cost of keeping everything in RAM + +Performance is only part of the equation. The infrastructure required to keep an online feature store entirely in memory can become a significant cost as the dataset grows. + +Consider a deployment with: + +* 50 million entities +* 100 online features per entity +* An average value size of 8 bytes + +That produces about 40 GB of raw feature data. + +Encoding overhead and replicas for high availability increase the provisioned footprint to roughly 120 GB. + +An all-in-memory Redis deployment would need to hold that entire footprint in RAM. With Aerospike HMA, the feature data can reside on SSD while only the index needs to remain in memory. For this example, the index requires only a few gigabytes of RAM. + +The advantage comes from how much memory each design needs: + +| Model | In RAM | On SSD | +| :---- | :---- | :---- | +| **Redis** (RAM) | \~120 GB | None | +| **Aerospike** (index RAM, data SSD) | \~6 GB | 120 GB | + +Redis holds the entire footprint in RAM. Aerospike keeps only the index there and puts the feature data on SSD. Because RAM costs more than fifty times as much per gigabyte as SSD, moving the bulk of the footprint off memory cuts the RAM you have to provision by roughly 20 times, and the total infrastructure cost by an order of magnitude. + +![Infrastructure cost with increasing scale: Aerospike versus Redis](/images/blog/aerospike-benchmark-infrastructure-cost.png) + +**Chart 5: Costs dramatically rise with entity count for an all-RAM architecture** + +The gap widens as the dataset grows, which is what the chart above shows. These figures are illustrative rather than a cloud-provider quote, so use your own entity counts, prices, and latency targets for planning. + +The difference grows as entity counts reach hundreds of millions or feature vectors become wider. It narrows when the entire working set is small enough to fit comfortably in RAM or when the workload requires every feature to have the lowest possible latency. + +Lower infrastructure cost only helps if it still meets the latency target. The benchmark results show that SSD-based feature storage does in fact deliver near in-memory performance for these Feast workloads. + +## Try it yourself + +For teams using Feast, Aerospike provides an alternative that combines the performance of an in-memory architecture with the capacity and economics of SSD-based storage. For controlled, head-to-head benchmarks run under our own methodology, see Aerospike's competitive research, including a recent [Redis comparison](https://aerospike.com/resources/benchmarks/aerospike-vs-redis-benchmark-report/) and [DynamoDB results](https://aerospike.com/resources/benchmarks/aerospike-dynamodb-benchmark/). + +For production planning, use your own entity counts, feature widths, request rates, and latency targets. The right online store depends on the shape of your specific workload as well as the total size of your dataset. + +Get started with the Aerospike online store for [Feast here](https://github.com/feast-dev/feast/blob/master/docs/reference/online-stores/aerospike.md). diff --git a/infra/website/docs/blog/aerospike-feast-now-available.md b/infra/website/docs/blog/aerospike-feast-now-available.md new file mode 100644 index 00000000000..9e44fbbd341 --- /dev/null +++ b/infra/website/docs/blog/aerospike-feast-now-available.md @@ -0,0 +1,78 @@ +--- +title: "Aerospike now available as a Feast online store" +description: "Feast now supports Aerospike as an online store, so teams can serve features with low, predictable latency while keeping feature data on SSD instead of RAM." +date: 2026-08-20 +authors: ["Valentyn Kahamlyk", "Francisco Javier Arceo"] +--- + +Feast, a popular open-source [feature store](https://aerospike.com/blog/feature-store), now supports Aerospike as an online store, giving ML teams a way to serve features with low, predictable latency as their feature sets and workloads grow. + +With Aerospike as the serving backend, Feast users can scale beyond the practical limits and cost of an entirely in-memory store while maintaining the performance required for online inference. + +Ready to try it? Jump to [getting started with Aerospike in Feast](#get-started-with-feast-and-aerospike). + +## Aerospike for online feature serving + +Customers like [Myntra](https://aerospike.com/resources/customer-stories/myntra/), [Sony](https://aerospike.com/resources/customer-stories/sony/), [MGID](https://aerospike.com/resources/customer-stories/mgid-aerospike-customer-story/), and [PhonePe](https://aerospike.com/resources/customer-stories/phonepe/) already use Aerospike to serve features for applications, such as [fraud detection](https://aerospike.com/solutions/use-cases/fraud-prevention/), personalization, and [real-time bidding](https://aerospike.com/solutions/industry/adtech/), where decisions need to be made in milliseconds as conditions change. The Feast integration lets customers connect feature views to Aerospike as the online store while continuing to manage definitions, the registry, and feature services in Feast. For existing Aerospike customers, this provides a straightforward way to connect Feast to a database they already use. For new deployments, Aerospike can serve as the online store from the beginning. + +## Performance without an all-in-memory architecture + +When selecting an online store for Feast, teams often start with an in-memory database. That works well when the feature set is small enough to fit comfortably in RAM. As the number of entities and features grows, however, keeping the entire dataset in memory can become increasingly expensive. + +Aerospike is built to provide in-memory performance with the scale benefits of SSD. With [Hybrid Memory Architecture (HMA)](https://aerospike.com/blog/hybrid-memory-architecture-optimization/), indexes remain in memory while feature data resides on SSD, so the database does not need to keep the entire dataset in RAM. The incremental latency of an SSD read is measured in microseconds, while the network round trip for an online feature request typically takes low milliseconds, so retrieving feature data from SSD adds little to the total response time. This gives teams a path to larger feature sets without keeping every feature value in RAM. The same design also holds up as concurrency rises, because keeping everything in RAM does not by itself guarantee that a store will remain responsive. + +## Benchmarking in Feast + +We ran Feast's `feast-benchmarks` harness end to end, from the load generator through `feast serve` to the online store, varying the number of entities per request, the number of features per request, and the request rate. Three results stood out: + +1. **On smaller requests, Aerospike matched the in-memory store**, serving single-entity and modest-batch requests in the same low-millisecond range. +2. **As load increased, Aerospike continued serving successfully** on demanding request shapes where the in-memory store began to degrade. +3. **Aerospike delivered that performance while storing feature data on SSD**, rather than requiring the entire feature set to remain in RAM. + +Full methodology and per-workload results are in [Fast like a cache, priced like storage: Benchmarking Aerospike on Feast](/blog/aerospike-feast-benchmark-harness-results). + +## Out of the box: Configuring Aerospike with Feast + +Feast supports a range of online stores through a common interface, so using Aerospike is a configuration choice rather than a code change. Aerospike works out of the box with HMA: indexes remain in memory while feature data resides on SSD. No additional Feast configuration is required to get that behavior. + +A basic configuration consists of the store type and a namespace: + +```yaml +online_store: + type: aerospike + namespace: feast_ssd +``` + +HMA is the default and is the right choice for most feature views. If a small number of feature views are especially latency-sensitive, you can route them to a memory-backed namespace: + +```yaml +online_store: + type: aerospike + namespace: feast_ssd + namespace_overrides: + scoring_velocity: feast_ram +``` + +This lets you use RAM selectively rather than making the entire feature store an in-memory deployment: + +* **HMA namespace:** index in RAM, feature data on SSD +* **Memory namespace:** index and feature data in RAM + +### A final performance tip + +To keep read times low, group feature views that are commonly read together on the same namespace. If a request pulls feature views from different namespaces, it requires a separate read from each. For wide feature services, you can also use `precompute_online=True` to combine features into a single lookup. + +See the [Online server performance tuning guide](https://github.com/feast-dev/feast/blob/master/docs/how-to-guides/online-server-performance-tuning.md) and [Aerospike online store reference](https://github.com/feast-dev/feast/blob/master/docs/reference/online-stores/aerospike.md) for the configuration details, including TTL, timeouts, set overrides, and prewriting hooks. + +## Get started with Feast and Aerospike + +The fastest way to start running the integration is directly through [Feast here](https://github.com/feast-dev/feast/blob/master/docs/reference/online-stores/aerospike.md). Pin your Feast version and test failover, TTL, and namespace placement in a staging cluster before moving to production. + +There are also several tutorials on using Aerospike as an online store: + +* [Aerospike Feature store tutorial](https://aerospike.com/docs/develop/tutorials/applications/feature-store/) +* [Serve real-time Feast features with Aerospike](https://aerospike.com/docs/develop/feast-aerospike-online-store) +* [Route hot and cold features across RAM and SSD](https://aerospike.com/docs/develop/feast-aerospike-tiering) +* [Stream fraud velocity features with Feast](https://aerospike.com/docs/develop/feast-aerospike-fraud-velocity) + +With the integration now available through Feast, teams can use Aerospike as the serving layer as their feature workloads grow. \ No newline at end of file diff --git a/infra/website/public/images/blog/aerospike-benchmark-infrastructure-cost.png b/infra/website/public/images/blog/aerospike-benchmark-infrastructure-cost.png new file mode 100644 index 00000000000..1e64184e89e Binary files /dev/null and b/infra/website/public/images/blog/aerospike-benchmark-infrastructure-cost.png differ diff --git a/infra/website/public/images/blog/aerospike-benchmark-p99-latency-entity-growth.png b/infra/website/public/images/blog/aerospike-benchmark-p99-latency-entity-growth.png new file mode 100644 index 00000000000..c2ed9513aa8 Binary files /dev/null and b/infra/website/public/images/blog/aerospike-benchmark-p99-latency-entity-growth.png differ diff --git a/infra/website/public/images/blog/aerospike-benchmark-p99-latency-feature-growth.png b/infra/website/public/images/blog/aerospike-benchmark-p99-latency-feature-growth.png new file mode 100644 index 00000000000..32ff5aeccaf Binary files /dev/null and b/infra/website/public/images/blog/aerospike-benchmark-p99-latency-feature-growth.png differ diff --git a/infra/website/public/images/blog/aerospike-benchmark-success-rate-rps-wide-requests.png b/infra/website/public/images/blog/aerospike-benchmark-success-rate-rps-wide-requests.png new file mode 100644 index 00000000000..191dc499a0a Binary files /dev/null and b/infra/website/public/images/blog/aerospike-benchmark-success-rate-rps-wide-requests.png differ diff --git a/infra/website/public/images/blog/aerospike-benchmark-success-rate-rps.png b/infra/website/public/images/blog/aerospike-benchmark-success-rate-rps.png new file mode 100644 index 00000000000..6c9f30cebfe Binary files /dev/null and b/infra/website/public/images/blog/aerospike-benchmark-success-rate-rps.png differ 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/sdk/python/tests/unit/online_store/test_online_retrieval.py b/sdk/python/tests/unit/online_store/test_online_retrieval.py index 96853618c59..d4b2867af6b 100644 --- a/sdk/python/tests/unit/online_store/test_online_retrieval.py +++ b/sdk/python/tests/unit/online_store/test_online_retrieval.py @@ -158,7 +158,9 @@ def test_get_online_features() -> None: # Feature values assert tensor_result["lon"] == ["1.0", "1.0"] # String -> not tensor - assert torch.equal(tensor_result["avg_orders_day"], torch.tensor([1.0, 1.0])) + assert torch.equal( + tensor_result["avg_orders_day"], torch.tensor([1.0, 1.0], device=device) + ) assert tensor_result["name"] == ["John", "John"] assert torch.equal(tensor_result["trips"], torch.tensor([7, 7], device=device)) 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,