diff --git a/.secrets.baseline b/.secrets.baseline index b815a28edc9..ca3288fa084 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -966,21 +966,21 @@ "filename": "infra/feast-operator/api/v1/zz_generated.deepcopy.go", "hashed_secret": "f914fc9324de1bec1ad13dec94a8ea2ddb41fc87", "is_verified": false, - "line_number": 817 + "line_number": 874 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/api/v1/zz_generated.deepcopy.go", "hashed_secret": "44e17306b837162269a410204daaa5ecee4ec22c", "is_verified": false, - "line_number": 878 + "line_number": 935 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/api/v1/zz_generated.deepcopy.go", "hashed_secret": "c2028031c154bbe86fd69bef740855c74b927dcf", "is_verified": false, - "line_number": 1570 + "line_number": 1627 } ], "infra/feast-operator/api/v1alpha1/featurestore_types.go": [ diff --git a/infra/feast-operator/api/v1/featurestore_types.go b/infra/feast-operator/api/v1/featurestore_types.go index 271ae1fe6b1..c023eda4e00 100644 --- a/infra/feast-operator/api/v1/featurestore_types.go +++ b/infra/feast-operator/api/v1/featurestore_types.go @@ -142,6 +142,46 @@ type OpenLineageConsumerConfig struct { NamespaceMapping map[string]string `json:"namespaceMapping,omitempty"` } +// MlflowConfig enables MLflow experiment tracking integration for Feast. +// When enabled, feature retrieval metadata is automatically logged to MLflow runs +// and the Feast UI displays lineage from feature views to registered models. +type MlflowConfig struct { + // Enable MLflow integration. + Enabled bool `json:"enabled"` + // MLflow tracking server URI. When omitted, the operator auto-discovers + // from the cluster MLflow CR (status.address.url). Falls back to + // MLFLOW_TRACKING_URI env var on pods. + // +optional + TrackingUri *string `json:"trackingUri,omitempty"` + // Automatically log feature metadata on every retrieval inside an active MLflow run. + // Defaults to true when enabled. + // +optional + AutoLog *bool `json:"autoLog,omitempty"` + // Save entity DataFrame as MLflow artifact on historical retrieval. + // Defaults to false. + // +optional + AutoLogEntityDf *bool `json:"autoLogEntityDf,omitempty"` + // Maximum number of entity DataFrame rows to save as an MLflow artifact. + // DataFrames exceeding this limit are skipped. Defaults to 100000. + // +kubebuilder:validation:Minimum=1 + // +optional + EntityDfMaxRows *int32 `json:"entityDfMaxRows,omitempty"` + // Log feast apply and materialize operations to a separate MLflow experiment. + // Defaults to false. + // +optional + LogOperations *bool `json:"logOperations,omitempty"` + // Suffix appended to the project name for the operations experiment. + // Defaults to "-feast-ops". + // +optional + OpsExperimentSuffix *string `json:"opsExperimentSuffix,omitempty"` + // ExtraConfig holds additional MLflow key-value settings written inline into + // the mlflow block of feature_store.yaml. Boolean and integer string values + // are coerced to native YAML types. Keys must be valid Feast MlflowConfig + // YAML field names. + // +optional + ExtraConfig map[string]string `json:"extraConfig,omitempty"` +} + // FeatureStoreSpec defines the desired state of FeatureStore // +kubebuilder:validation:XValidation:rule="self.replicas <= 1 || !has(self.services) || !has(self.services.scaling) || !has(self.services.scaling.autoscaling)",message="replicas > 1 and services.scaling.autoscaling are mutually exclusive." // +kubebuilder:validation:XValidation:rule="self.replicas <= 1 && (!has(self.services) || !has(self.services.scaling) || !has(self.services.scaling.autoscaling)) || (has(self.services) && has(self.services.onlineStore) && has(self.services.onlineStore.persistence) && has(self.services.onlineStore.persistence.store))",message="Scaling requires DB-backed persistence for the online store. Configure services.onlineStore.persistence.store when using replicas > 1 or autoscaling." @@ -172,6 +212,12 @@ type FeatureStoreSpec struct { // Written into feature_store.yaml for all service pods. // +optional OpenLineage *OpenLineageConfig `json:"openlineage,omitempty"` + // Mlflow enables MLflow experiment tracking integration for Feast. + // Written into feature_store.yaml for all service pods and the client ConfigMap. + // When omitted and a cluster MLflow instance is detected, defaults to enabled + // with the discovered tracking URI. + // +optional + Mlflow *MlflowConfig `json:"mlflow,omitempty"` } // FeastProjectDir defines how to create the feast project directory. diff --git a/infra/feast-operator/api/v1/zz_generated.deepcopy.go b/infra/feast-operator/api/v1/zz_generated.deepcopy.go index 2a6b6a69266..b3edfc05387 100644 --- a/infra/feast-operator/api/v1/zz_generated.deepcopy.go +++ b/infra/feast-operator/api/v1/zz_generated.deepcopy.go @@ -497,6 +497,11 @@ func (in *FeatureStoreSpec) DeepCopyInto(out *FeatureStoreSpec) { *out = new(OpenLineageConfig) (*in).DeepCopyInto(*out) } + if in.Mlflow != nil { + in, out := &in.Mlflow, &out.Mlflow + *out = new(MlflowConfig) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FeatureStoreSpec. @@ -761,6 +766,58 @@ func (in *McpConfig) DeepCopy() *McpConfig { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *MlflowConfig) DeepCopyInto(out *MlflowConfig) { + *out = *in + if in.TrackingUri != nil { + in, out := &in.TrackingUri, &out.TrackingUri + *out = new(string) + **out = **in + } + if in.AutoLog != nil { + in, out := &in.AutoLog, &out.AutoLog + *out = new(bool) + **out = **in + } + if in.AutoLogEntityDf != nil { + in, out := &in.AutoLogEntityDf, &out.AutoLogEntityDf + *out = new(bool) + **out = **in + } + if in.EntityDfMaxRows != nil { + in, out := &in.EntityDfMaxRows, &out.EntityDfMaxRows + *out = new(int32) + **out = **in + } + if in.LogOperations != nil { + in, out := &in.LogOperations, &out.LogOperations + *out = new(bool) + **out = **in + } + if in.OpsExperimentSuffix != nil { + in, out := &in.OpsExperimentSuffix, &out.OpsExperimentSuffix + *out = new(string) + **out = **in + } + if in.ExtraConfig != nil { + in, out := &in.ExtraConfig, &out.ExtraConfig + *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 MlflowConfig. +func (in *MlflowConfig) DeepCopy() *MlflowConfig { + if in == nil { + return nil + } + out := new(MlflowConfig) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *OfflinePushBatchingConfig) DeepCopyInto(out *OfflinePushBatchingConfig) { *out = *in diff --git a/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml b/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml index 19af99046e9..d4f8e03c8c8 100644 --- a/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml +++ b/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml @@ -147,7 +147,7 @@ metadata: } ] capabilities: Basic Install - createdAt: "2026-07-20T13:27:58Z" + createdAt: "2026-07-16T11:41:33Z" operators.operatorframework.io/builder: operator-sdk-v1.41.0 operators.operatorframework.io/project_layout: go.kubebuilder.io/v4 name: feast-operator.v0.65.0 @@ -307,6 +307,14 @@ spec: - get - patch - update + - apiGroups: + - mlflow.opendatahub.io + resources: + - mlflows + verbs: + - get + - list + - watch - apiGroups: - monitoring.coreos.com resources: diff --git a/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml b/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml index 338dc75bd63..b342e4a3363 100644 --- a/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml +++ b/infra/feast-operator/bundle/manifests/feast.dev_featurestores.yaml @@ -834,6 +834,56 @@ spec: minimum: 1 type: integer type: object + mlflow: + description: |- + Mlflow enables MLflow experiment tracking integration for Feast. + Written into feature_store. + properties: + autoLog: + description: |- + Automatically log feature metadata on every retrieval inside an active MLflow run. + Defaults to true when enabled. + type: boolean + autoLogEntityDf: + description: |- + Save entity DataFrame as MLflow artifact on historical retrieval. + Defaults to false. + type: boolean + enabled: + description: Enable MLflow integration. + type: boolean + entityDfMaxRows: + description: |- + Maximum number of entity DataFrame rows to save as an MLflow artifact. + DataFrames exceeding this limit are skipped. + format: int32 + minimum: 1 + type: integer + extraConfig: + additionalProperties: + type: string + description: |- + ExtraConfig holds additional MLflow key-value settings written inline into + the mlflow block of feature_store.yaml. + type: object + logOperations: + description: |- + Log feast apply and materialize operations to a separate MLflow experiment. + Defaults to false. + type: boolean + opsExperimentSuffix: + description: |- + Suffix appended to the project name for the operations experiment. + Defaults to "-feast-ops". + type: string + trackingUri: + description: |- + MLflow tracking server URI. When omitted, the operator auto-discovers + from the cluster MLflow CR (status.address.url). + type: string + required: + - enabled + type: object openlineage: description: |- OpenLineage enables OpenLineage data lineage tracking for Feast operations. @@ -7123,6 +7173,56 @@ spec: minimum: 1 type: integer type: object + mlflow: + description: |- + Mlflow enables MLflow experiment tracking integration for Feast. + Written into feature_store. + properties: + autoLog: + description: |- + Automatically log feature metadata on every retrieval inside an active MLflow run. + Defaults to true when enabled. + type: boolean + autoLogEntityDf: + description: |- + Save entity DataFrame as MLflow artifact on historical retrieval. + Defaults to false. + type: boolean + enabled: + description: Enable MLflow integration. + type: boolean + entityDfMaxRows: + description: |- + Maximum number of entity DataFrame rows to save as an MLflow artifact. + DataFrames exceeding this limit are skipped. + format: int32 + minimum: 1 + type: integer + extraConfig: + additionalProperties: + type: string + description: |- + ExtraConfig holds additional MLflow key-value settings written inline into + the mlflow block of feature_store.yaml. + type: object + logOperations: + description: |- + Log feast apply and materialize operations to a separate MLflow experiment. + Defaults to false. + type: boolean + opsExperimentSuffix: + description: |- + Suffix appended to the project name for the operations experiment. + Defaults to "-feast-ops". + type: string + trackingUri: + description: |- + MLflow tracking server URI. When omitted, the operator auto-discovers + from the cluster MLflow CR (status.address.url). + type: string + required: + - enabled + type: object openlineage: description: |- OpenLineage enables OpenLineage data lineage tracking for Feast operations. 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 b840d46a3b1..a781862e6cd 100644 --- a/infra/feast-operator/config/crd/bases/feast.dev_featurestores.yaml +++ b/infra/feast-operator/config/crd/bases/feast.dev_featurestores.yaml @@ -834,6 +834,56 @@ spec: minimum: 1 type: integer type: object + mlflow: + description: |- + Mlflow enables MLflow experiment tracking integration for Feast. + Written into feature_store. + properties: + autoLog: + description: |- + Automatically log feature metadata on every retrieval inside an active MLflow run. + Defaults to true when enabled. + type: boolean + autoLogEntityDf: + description: |- + Save entity DataFrame as MLflow artifact on historical retrieval. + Defaults to false. + type: boolean + enabled: + description: Enable MLflow integration. + type: boolean + entityDfMaxRows: + description: |- + Maximum number of entity DataFrame rows to save as an MLflow artifact. + DataFrames exceeding this limit are skipped. + format: int32 + minimum: 1 + type: integer + extraConfig: + additionalProperties: + type: string + description: |- + ExtraConfig holds additional MLflow key-value settings written inline into + the mlflow block of feature_store.yaml. + type: object + logOperations: + description: |- + Log feast apply and materialize operations to a separate MLflow experiment. + Defaults to false. + type: boolean + opsExperimentSuffix: + description: |- + Suffix appended to the project name for the operations experiment. + Defaults to "-feast-ops". + type: string + trackingUri: + description: |- + MLflow tracking server URI. When omitted, the operator auto-discovers + from the cluster MLflow CR (status.address.url). + type: string + required: + - enabled + type: object openlineage: description: |- OpenLineage enables OpenLineage data lineage tracking for Feast operations. @@ -7123,6 +7173,56 @@ spec: minimum: 1 type: integer type: object + mlflow: + description: |- + Mlflow enables MLflow experiment tracking integration for Feast. + Written into feature_store. + properties: + autoLog: + description: |- + Automatically log feature metadata on every retrieval inside an active MLflow run. + Defaults to true when enabled. + type: boolean + autoLogEntityDf: + description: |- + Save entity DataFrame as MLflow artifact on historical retrieval. + Defaults to false. + type: boolean + enabled: + description: Enable MLflow integration. + type: boolean + entityDfMaxRows: + description: |- + Maximum number of entity DataFrame rows to save as an MLflow artifact. + DataFrames exceeding this limit are skipped. + format: int32 + minimum: 1 + type: integer + extraConfig: + additionalProperties: + type: string + description: |- + ExtraConfig holds additional MLflow key-value settings written inline into + the mlflow block of feature_store.yaml. + type: object + logOperations: + description: |- + Log feast apply and materialize operations to a separate MLflow experiment. + Defaults to false. + type: boolean + opsExperimentSuffix: + description: |- + Suffix appended to the project name for the operations experiment. + Defaults to "-feast-ops". + type: string + trackingUri: + description: |- + MLflow tracking server URI. When omitted, the operator auto-discovers + from the cluster MLflow CR (status.address.url). + type: string + required: + - enabled + type: object openlineage: description: |- OpenLineage enables OpenLineage data lineage tracking for Feast operations. diff --git a/infra/feast-operator/config/rbac/role.yaml b/infra/feast-operator/config/rbac/role.yaml index a79dca283ed..61500bc5d30 100644 --- a/infra/feast-operator/config/rbac/role.yaml +++ b/infra/feast-operator/config/rbac/role.yaml @@ -136,6 +136,14 @@ rules: - get - patch - update +- apiGroups: + - mlflow.opendatahub.io + resources: + - mlflows + verbs: + - get + - list + - watch - apiGroups: - monitoring.coreos.com resources: diff --git a/infra/feast-operator/dist/install.yaml b/infra/feast-operator/dist/install.yaml index 854d625bff5..a2f0e28f14b 100644 --- a/infra/feast-operator/dist/install.yaml +++ b/infra/feast-operator/dist/install.yaml @@ -842,6 +842,56 @@ spec: minimum: 1 type: integer type: object + mlflow: + description: |- + Mlflow enables MLflow experiment tracking integration for Feast. + Written into feature_store. + properties: + autoLog: + description: |- + Automatically log feature metadata on every retrieval inside an active MLflow run. + Defaults to true when enabled. + type: boolean + autoLogEntityDf: + description: |- + Save entity DataFrame as MLflow artifact on historical retrieval. + Defaults to false. + type: boolean + enabled: + description: Enable MLflow integration. + type: boolean + entityDfMaxRows: + description: |- + Maximum number of entity DataFrame rows to save as an MLflow artifact. + DataFrames exceeding this limit are skipped. + format: int32 + minimum: 1 + type: integer + extraConfig: + additionalProperties: + type: string + description: |- + ExtraConfig holds additional MLflow key-value settings written inline into + the mlflow block of feature_store.yaml. + type: object + logOperations: + description: |- + Log feast apply and materialize operations to a separate MLflow experiment. + Defaults to false. + type: boolean + opsExperimentSuffix: + description: |- + Suffix appended to the project name for the operations experiment. + Defaults to "-feast-ops". + type: string + trackingUri: + description: |- + MLflow tracking server URI. When omitted, the operator auto-discovers + from the cluster MLflow CR (status.address.url). + type: string + required: + - enabled + type: object openlineage: description: |- OpenLineage enables OpenLineage data lineage tracking for Feast operations. @@ -7131,6 +7181,56 @@ spec: minimum: 1 type: integer type: object + mlflow: + description: |- + Mlflow enables MLflow experiment tracking integration for Feast. + Written into feature_store. + properties: + autoLog: + description: |- + Automatically log feature metadata on every retrieval inside an active MLflow run. + Defaults to true when enabled. + type: boolean + autoLogEntityDf: + description: |- + Save entity DataFrame as MLflow artifact on historical retrieval. + Defaults to false. + type: boolean + enabled: + description: Enable MLflow integration. + type: boolean + entityDfMaxRows: + description: |- + Maximum number of entity DataFrame rows to save as an MLflow artifact. + DataFrames exceeding this limit are skipped. + format: int32 + minimum: 1 + type: integer + extraConfig: + additionalProperties: + type: string + description: |- + ExtraConfig holds additional MLflow key-value settings written inline into + the mlflow block of feature_store.yaml. + type: object + logOperations: + description: |- + Log feast apply and materialize operations to a separate MLflow experiment. + Defaults to false. + type: boolean + opsExperimentSuffix: + description: |- + Suffix appended to the project name for the operations experiment. + Defaults to "-feast-ops". + type: string + trackingUri: + description: |- + MLflow tracking server URI. When omitted, the operator auto-discovers + from the cluster MLflow CR (status.address.url). + type: string + required: + - enabled + type: object openlineage: description: |- OpenLineage enables OpenLineage data lineage tracking for Feast operations. @@ -22164,6 +22264,14 @@ rules: - get - patch - update +- apiGroups: + - mlflow.opendatahub.io + resources: + - mlflows + verbs: + - get + - list + - watch - apiGroups: - monitoring.coreos.com resources: diff --git a/infra/feast-operator/docs/api/markdown/ref.md b/infra/feast-operator/docs/api/markdown/ref.md index 0a7782feb2a..80118ab5fd0 100644 --- a/infra/feast-operator/docs/api/markdown/ref.md +++ b/infra/feast-operator/docs/api/markdown/ref.md @@ -300,6 +300,10 @@ Mutually exclusive with services.scaling.autoscaling. | Written into feature_store.yaml for all service pods. | | `openlineage` _[OpenLineageConfig](#openlineageconfig)_ | OpenLineage enables OpenLineage data lineage tracking for Feast operations. Written into feature_store.yaml for all service pods. | +| `mlflow` _[MlflowConfig](#mlflowconfig)_ | Mlflow enables MLflow experiment tracking integration for Feast. +Written into feature_store.yaml for all service pods and the client ConfigMap. +When omitted and a cluster MLflow instance is detected, defaults to enabled +with the discovered tracking URI. | #### FeatureStoreStatus @@ -526,6 +530,39 @@ _Appears in:_ | `transport` _string_ | MCP transport protocol. | +#### MlflowConfig + + + +MlflowConfig enables MLflow experiment tracking integration for Feast. +When enabled, feature retrieval metadata is automatically logged to MLflow runs +and the Feast UI displays lineage from feature views to registered models. + +_Appears in:_ +- [FeatureStoreSpec](#featurestorespec) + +| Field | Description | +| --- | --- | +| `enabled` _boolean_ | Enable MLflow integration. | +| `trackingUri` _string_ | MLflow tracking server URI. When omitted, the operator auto-discovers +from the cluster MLflow CR (status.address.url). Falls back to +MLFLOW_TRACKING_URI env var on pods. | +| `autoLog` _boolean_ | Automatically log feature metadata on every retrieval inside an active MLflow run. +Defaults to true when enabled. | +| `autoLogEntityDf` _boolean_ | Save entity DataFrame as MLflow artifact on historical retrieval. +Defaults to false. | +| `entityDfMaxRows` _integer_ | Maximum number of entity DataFrame rows to save as an MLflow artifact. +DataFrames exceeding this limit are skipped. Defaults to 100000. | +| `logOperations` _boolean_ | Log feast apply and materialize operations to a separate MLflow experiment. +Defaults to false. | +| `opsExperimentSuffix` _string_ | Suffix appended to the project name for the operations experiment. +Defaults to "-feast-ops". | +| `extraConfig` _object (keys:string, values:string)_ | ExtraConfig holds additional MLflow key-value settings written inline into +the mlflow block of feature_store.yaml. Boolean and integer string values +are coerced to native YAML types. Keys must be valid Feast MlflowConfig +YAML field names. | + + #### OfflinePushBatchingConfig diff --git a/infra/feast-operator/docs/mlflow-integration.md b/infra/feast-operator/docs/mlflow-integration.md new file mode 100644 index 00000000000..e8c8b9c26f7 --- /dev/null +++ b/infra/feast-operator/docs/mlflow-integration.md @@ -0,0 +1,133 @@ +# MLflow Integration (RHOAI / ODH) + +## Overview + +When both the Feast operator and MLflow operator are enabled (`Managed`) on RHOAI/ODH, the Feast operator automatically detects the cluster MLflow instance and enables MLflow experiment tracking for every FeatureStore deployment. This provides: + +- **Zero-config MLflow lineage** for workbench users (no YAML editing) +- **Feast UI lineage panels** (training runs, model associations, registry graph) via the operator-managed UI Route +- **Operations audit trail** for `feast apply` and `feast materialize` + +## How it works + +1. The Feast operator looks for the singleton cluster-scoped `MLflow` CR (`mlflow.opendatahub.io/v1`, name `mlflow`). +2. If found and Ready (checked via `status.conditions`), the operator reads `status.address.url` (in-cluster HTTPS) as the tracking URI. +3. The `mlflow:` block is written into: + - **Server `feature_store.yaml`** (all Feast pods: online, offline, registry, UI, cron) + - **Client ConfigMap** (workbench notebooks) +4. `MLFLOW_TRACKING_AUTH=kubernetes-namespaced` and `MLFLOW_TRACKING_URI` env vars are injected into Feast service containers. +5. An `mlflow-integration` RoleBinding is created for the FeatureStore ServiceAccount. +6. The controller watches the MLflow CR, so FeatureStores are automatically reconciled when MLflow becomes Ready or is removed. + +## FeatureStore CR configuration + +### Auto-enabled (default when MLflow is present) + +No `spec.mlflow` needed. The operator auto-enables when MLflow is detected: + +```yaml +apiVersion: feast.dev/v1 +kind: FeatureStore +metadata: + name: my-store +spec: + feastProject: my_project + services: + onlineStore: {} + registry: {} + ui: {} + # mlflow is auto-enabled — no config required +``` + +### Explicit configuration + +Override defaults or enable additional features: + +```yaml +apiVersion: feast.dev/v1 +kind: FeatureStore +metadata: + name: my-store +spec: + feastProject: my_project + services: + onlineStore: {} + registry: {} + ui: {} + mlflow: + enabled: true + trackingUri: "https://custom-mlflow.example.com:8443" + autoLog: true + autoLogEntityDf: true + entityDfMaxRows: 50000 + logOperations: true + opsExperimentSuffix: "-feast-ops" +``` + +### Opt-out + +Disable MLflow even when the MLflow operator is present: + +```yaml +spec: + mlflow: + enabled: false +``` + +## Configuration options + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `enabled` | bool | auto-detected | Master switch for MLflow integration | +| `trackingUri` | string | auto-discovered | MLflow tracking server URI | +| `autoLog` | *bool | `true` | Auto-log feature metadata on every retrieval | +| `autoLogEntityDf` | *bool | `false` | Save entity DataFrame as artifact | +| `entityDfMaxRows` | *int32 | `100000` | Skip artifact for large DataFrames | +| `logOperations` | *bool | `false` | Log `feast apply` / `materialize` to ops experiment | +| `opsExperimentSuffix` | *string | `"-feast-ops"` | Ops experiment name suffix | +| `extraConfig` | map[string]string | — | Additional YAML fields (coerced to native types) | + +## Where to see lineage + +Lineage appears in the **Feast UI** (not the RHOAI Dashboard): + +1. Open the Feast UI Route: `oc get route -l app.kubernetes.io/name=` +2. Feature View page shows MLflow training run count, last used, and model associations. +3. Registry visualization draws edges from FeatureService through MLflow runs to registered models. +4. Click model links to open the RHOAI MLflow UI for that run/model. + +The RHOAI Dashboard provides navigation to both the Feast UI and the MLflow application tile. + +## Workbench setup + +For workbench notebooks to use `store.mlflow`: + +1. Select the Feast project in the RHOAI Dashboard (mounts client ConfigMap). +2. Ensure the workbench has the `opendatahub.io/mlflow-instance` annotation (enables MLflow SDK env injection by the MLflow operator). + +Then in the notebook: + +```python +from feast import FeatureStore + +store = FeatureStore(...) # from mounted client config + +with store.mlflow.start_run(run_name="training"): + df = store.get_historical_features(...).to_df() + model = train(df) + store.mlflow.log_model(model, "model") +``` + +## Tracking URI resolution order + +1. Explicit `trackingUri` in FeatureStore CR +2. Auto-discovered from MLflow CR `status.address.url` +3. `MLFLOW_TRACKING_URI` environment variable (on workbench pods, injected by MLflow operator) +4. MLflow default (`./mlruns`) + +## Graceful degradation + +- If MLflow operator is not installed: no mlflow block in YAML; FeatureStore stays Ready. +- If MLflow CR is not Ready: discovery returns empty; mlflow stays off. +- If tracking URI becomes unreachable: SDK logs a warning but does not block feature retrieval. +- If UI pod lacks RBAC: `/api/mlflow-*` returns empty responses; lineage panels are hidden. diff --git a/infra/feast-operator/internal/controller/featurestore_controller.go b/infra/feast-operator/internal/controller/featurestore_controller.go index b94808c0df5..6f73dfa7958 100644 --- a/infra/feast-operator/internal/controller/featurestore_controller.go +++ b/infra/feast-operator/internal/controller/featurestore_controller.go @@ -79,6 +79,7 @@ type FeatureStoreReconciler struct { // +kubebuilder:rbac:groups=autoscaling,resources=horizontalpodautoscalers,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=monitoring.coreos.com,resources=servicemonitors,verbs=get;list;watch;create;patch;delete +// +kubebuilder:rbac:groups=mlflow.opendatahub.io,resources=mlflows,verbs=get;list;watch // Reconcile is part of the main kubernetes reconciliation loop which aims to // move the current state of the cluster closer to the desired state. @@ -272,6 +273,15 @@ func (r *FeatureStoreReconciler) SetupWithManager(mgr ctrl.Manager) error { }) bldr = bldr.Owns(sm) } + if services.HasMlflowCRD() { + mlflow := &unstructured.Unstructured{} + mlflow.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "mlflow.opendatahub.io", + Version: "v1", + Kind: "MLflow", + }) + bldr = bldr.Watches(mlflow, handler.EnqueueRequestsFromMapFunc(r.mapMlflowToFeastRequests)) + } return bldr.Complete(r) @@ -291,6 +301,24 @@ func (r *FeatureStoreReconciler) cleanupNamespaceRegistry(ctx context.Context, c return feast.RemoveFromNamespaceRegistry() } +// mapMlflowToFeastRequests re-queues every FeatureStore when the cluster MLflow +// CR changes so auto-discovery can pick up a newly Ready (or removed) instance. +func (r *FeatureStoreReconciler) mapMlflowToFeastRequests(ctx context.Context, _ client.Object) []reconcile.Request { + logger := log.FromContext(ctx) + var feastList feastdevv1.FeatureStoreList + if err := r.List(ctx, &feastList, client.InNamespace("")); err != nil { + logger.Error(err, "could not list FeatureStores for MLflow watch") + return nil + } + requests := make([]reconcile.Request, 0, len(feastList.Items)) + for _, obj := range feastList.Items { + requests = append(requests, reconcile.Request{ + NamespacedName: client.ObjectKeyFromObject(&obj), + }) + } + return requests +} + // if a remotely referenced FeatureStore is changed, reconcile any FeatureStores that reference it. func (r *FeatureStoreReconciler) mapFeastRefsToFeastRequests(ctx context.Context, object client.Object) []reconcile.Request { logger := log.FromContext(ctx) diff --git a/infra/feast-operator/internal/controller/services/mlflow_discover.go b/infra/feast-operator/internal/controller/services/mlflow_discover.go new file mode 100644 index 00000000000..6317f4d2cf8 --- /dev/null +++ b/infra/feast-operator/internal/controller/services/mlflow_discover.go @@ -0,0 +1,137 @@ +/* +Copyright 2024 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 services + +import ( + "context" + + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/log" +) + +var mlflowGVK = schema.GroupVersionKind{ + Group: "mlflow.opendatahub.io", + Version: "v1", + Kind: "MLflow", +} + +// DiscoverMlflowTrackingUri attempts to find the cluster-scoped MLflow CR +// (singleton named "mlflow") and returns its in-cluster tracking URI. +// Returns ("", false) when MLflow is not installed, not ready, or not available. +// This function never returns an error — it is designed for best-effort discovery +// so that the FeatureStore reconcile is not blocked by MLflow absence. +func DiscoverMlflowTrackingUri(ctx context.Context, c client.Client) (string, bool) { + mlflow := &unstructured.Unstructured{} + mlflow.SetGroupVersionKind(mlflowGVK) + + if err := c.Get(ctx, client.ObjectKey{Name: "mlflow"}, mlflow); err != nil { + return "", false + } + + status, found, _ := unstructured.NestedMap(mlflow.Object, "status") + if !found { + return "", false + } + + if !isMlflowReady(status) { + return "", false + } + + // Prefer in-cluster address (HTTPS service URL) + if addr, ok := status["address"].(map[string]interface{}); ok { + if url, ok := addr["url"].(string); ok && url != "" { + return url, true + } + } + + // Fallback to gateway URL (external, browser-reachable) + if url, ok := status["url"].(string); ok && url != "" { + return url, true + } + + return "", false +} + +// isMlflowReady checks status.conditions for a Ready=True condition. +// Returns true when conditions are absent (best-effort: older MLflow CRs may +// not report conditions, so we fall through to URL-based discovery). +func isMlflowReady(status map[string]interface{}) bool { + conditions, ok := status["conditions"].([]interface{}) + if !ok || len(conditions) == 0 { + return true + } + for _, c := range conditions { + cond, ok := c.(map[string]interface{}) + if !ok { + continue + } + condType, _ := cond["type"].(string) + condStatus, _ := cond["status"].(string) + if condType == "Ready" { + return condStatus == "True" + } + } + return true +} + +const mlflowRoleBindingSuffix = "-mlflow-integration" + +// deployMlflowRoleBinding creates or updates a RoleBinding that grants the +// FeatureStore's ServiceAccount access to the MLflow API via the +// mlflow-integration ClusterRole. This is required for Feast UI and service +// pods to query MLflow for lineage data. +// When MLflow is disabled, the RoleBinding is removed. +func (feast *FeastServices) deployMlflowRoleBinding() error { + applied := feast.Handler.FeatureStore.Status.Applied.Mlflow + rbName := GetFeastName(feast.Handler.FeatureStore) + mlflowRoleBindingSuffix + + rb := &rbacv1.RoleBinding{} + rb.Name = rbName + rb.Namespace = feast.Handler.FeatureStore.Namespace + rb.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("RoleBinding")) + + if applied == nil || !applied.Enabled { + return feast.Handler.DeleteOwnedFeastObj(rb) + } + + logger := log.FromContext(feast.Handler.Context) + if op, err := controllerutil.CreateOrUpdate(feast.Handler.Context, feast.Handler.Client, rb, controllerutil.MutateFn(func() error { + rb.Labels = feast.getLabels() + rb.Subjects = []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Name: GetFeastName(feast.Handler.FeatureStore), + Namespace: feast.Handler.FeatureStore.Namespace, + }, + } + rb.RoleRef = rbacv1.RoleRef{ + APIGroup: "rbac.authorization.k8s.io", + Kind: "ClusterRole", + Name: "mlflow-integration", + } + return controllerutil.SetControllerReference(feast.Handler.FeatureStore, rb, feast.Handler.Scheme) + })); err != nil { + return err + } else if op == controllerutil.OperationResultCreated || op == controllerutil.OperationResultUpdated { + logger.Info("Successfully reconciled MLflow RoleBinding", "RoleBinding", rbName, "operation", op) + } + return nil +} diff --git a/infra/feast-operator/internal/controller/services/repo_config.go b/infra/feast-operator/internal/controller/services/repo_config.go index 5df6bba3fbc..0468379692f 100644 --- a/infra/feast-operator/internal/controller/services/repo_config.go +++ b/infra/feast-operator/internal/controller/services/repo_config.go @@ -106,6 +106,10 @@ func getServiceRepoConfig( } } + if appliedSpec.Mlflow != nil && appliedSpec.Mlflow.Enabled { + setRepoConfigMlflow(appliedSpec.Mlflow, &repoConfig) + } + if appliedSpec.DataQualityMonitoring != nil { setRepoConfigDataQualityMonitoring(appliedSpec.DataQualityMonitoring, &repoConfig) } @@ -545,6 +549,27 @@ func coerceStringToYamlType(v string) interface{} { return v } +// setRepoConfigMlflow maps the CRD MlflowConfig into the mlflow YAML block. +func setRepoConfigMlflow(mlflow *feastdevv1.MlflowConfig, repoConfig *RepoConfig) { + yamlCfg := &MlflowYamlConfig{ + Enabled: mlflow.Enabled, + TrackingUri: mlflow.TrackingUri, + AutoLog: mlflow.AutoLog, + AutoLogEntityDf: mlflow.AutoLogEntityDf, + EntityDfMaxRows: mlflow.EntityDfMaxRows, + LogOperations: mlflow.LogOperations, + OpsExperimentSuffix: mlflow.OpsExperimentSuffix, + } + if len(mlflow.ExtraConfig) > 0 { + ec := make(map[string]interface{}, len(mlflow.ExtraConfig)) + for k, v := range mlflow.ExtraConfig { + ec[k] = coerceStringToYamlType(v) + } + yamlCfg.ExtraConfig = ec + } + repoConfig.Mlflow = yamlCfg +} + func setRepoConfigDataQualityMonitoring(dqmConfig *feastdevv1.DataQualityMonitoringConfig, repoConfig *RepoConfig) { if dqmConfig.AutoBaseline == nil { return @@ -602,6 +627,10 @@ func getClientRepoConfig( } } + if status.Applied.Mlflow != nil && status.Applied.Mlflow.Enabled { + setRepoConfigMlflow(status.Applied.Mlflow, &clientRepoConfig) + } + return clientRepoConfig } diff --git a/infra/feast-operator/internal/controller/services/repo_config_test.go b/infra/feast-operator/internal/controller/services/repo_config_test.go index 89941c6beb5..0423883593d 100644 --- a/infra/feast-operator/internal/controller/services/repo_config_test.go +++ b/infra/feast-operator/internal/controller/services/repo_config_test.go @@ -26,6 +26,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" feastdevv1 "github.com/feast-dev/feast/infra/feast-operator/api/v1" + handler "github.com/feast-dev/feast/infra/feast-operator/internal/controller/handler" ) var projectName = "test-project" @@ -1039,3 +1040,285 @@ var _ = Describe("TLS Certificate Path Configuration", func() { }) }) }) + +var _ = Describe("MLflow Configuration", func() { + Context("in getServiceRepoConfig", func() { + It("should set mlflow block with enabled + tracking URI", func() { + featureStore := minimalFeatureStore() + trackingUri := "https://mlflow.redhat-ods-applications.svc:8443" + autoLog := true + autoLogEntityDf := false + logOps := true + + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &trackingUri, + AutoLog: &autoLog, + AutoLogEntityDf: &autoLogEntityDf, + LogOperations: &logOps, + } + ApplyDefaultsToStatus(featureStore) + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, emptyMockExtractConfigFromConfigMap, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.Mlflow).NotTo(BeNil()) + Expect(repoConfig.Mlflow.Enabled).To(BeTrue()) + Expect(repoConfig.Mlflow.TrackingUri).To(Equal(&trackingUri)) + Expect(repoConfig.Mlflow.AutoLog).To(Equal(&autoLog)) + Expect(repoConfig.Mlflow.AutoLogEntityDf).To(Equal(&autoLogEntityDf)) + Expect(repoConfig.Mlflow.LogOperations).To(Equal(&logOps)) + }) + + It("should set mlflow block with entityDfMaxRows and opsExperimentSuffix", func() { + featureStore := minimalFeatureStore() + trackingUri := "https://mlflow.svc:8443" + maxRows := int32(5000) + suffix := "-my-ops" + + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &trackingUri, + EntityDfMaxRows: &maxRows, + OpsExperimentSuffix: &suffix, + } + ApplyDefaultsToStatus(featureStore) + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, emptyMockExtractConfigFromConfigMap, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.Mlflow).NotTo(BeNil()) + Expect(repoConfig.Mlflow.EntityDfMaxRows).To(Equal(&maxRows)) + Expect(repoConfig.Mlflow.OpsExperimentSuffix).To(Equal(&suffix)) + }) + + It("should set mlflow block with ExtraConfig coercing booleans", func() { + featureStore := minimalFeatureStore() + trackingUri := "https://mlflow.svc:8443" + + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &trackingUri, + ExtraConfig: map[string]string{ + "auto_log": stringTrue, + "auto_log_entity_df": "false", + }, + } + ApplyDefaultsToStatus(featureStore) + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, emptyMockExtractConfigFromConfigMap, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.Mlflow).NotTo(BeNil()) + Expect(repoConfig.Mlflow.ExtraConfig).To(HaveKeyWithValue("auto_log", true)) + Expect(repoConfig.Mlflow.ExtraConfig).To(HaveKeyWithValue("auto_log_entity_df", false)) + }) + + It("should not set mlflow block when spec.mlflow is nil", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = nil + ApplyDefaultsToStatus(featureStore) + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, emptyMockExtractConfigFromConfigMap, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.Mlflow).To(BeNil()) + }) + + It("should not set mlflow block when spec.mlflow.enabled is false", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: false, + } + ApplyDefaultsToStatus(featureStore) + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, emptyMockExtractConfigFromConfigMap, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.Mlflow).To(BeNil()) + }) + }) + + Context("in getClientRepoConfig", func() { + It("should include mlflow block in client config when enabled", func() { + featureStore := minimalFeatureStore() + trackingUri := "https://mlflow.svc:8443" + + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &trackingUri, + } + ApplyDefaultsToStatus(featureStore) + + repoConfig := getClientRepoConfig(featureStore, nil) + Expect(repoConfig.Mlflow).NotTo(BeNil()) + Expect(repoConfig.Mlflow.Enabled).To(BeTrue()) + Expect(repoConfig.Mlflow.TrackingUri).To(Equal(&trackingUri)) + }) + + It("should not include mlflow block in client config when disabled", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: false, + } + ApplyDefaultsToStatus(featureStore) + + repoConfig := getClientRepoConfig(featureStore, nil) + Expect(repoConfig.Mlflow).To(BeNil()) + }) + }) + + Context("applyMlflowDefaults", func() { + It("should clear applied mlflow when spec.mlflow.enabled is false", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{Enabled: false} + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + feast.applyMlflowDefaults() + Expect(featureStore.Status.Applied.Mlflow).To(BeNil()) + }) + + It("should keep spec values when spec.mlflow is explicitly set with trackingUri", func() { + featureStore := minimalFeatureStore() + trackingUri := "https://my-mlflow.svc:8443" + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &trackingUri, + } + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + feast.applyMlflowDefaults() + Expect(featureStore.Status.Applied.Mlflow).NotTo(BeNil()) + Expect(featureStore.Status.Applied.Mlflow.Enabled).To(BeTrue()) + Expect(featureStore.Status.Applied.Mlflow.TrackingUri).To(Equal(&trackingUri)) + }) + }) + + Context("isMlflowReady", func() { + It("should return true when no conditions are present", func() { + status := map[string]interface{}{} + Expect(isMlflowReady(status)).To(BeTrue()) + }) + + It("should return true when Ready condition is True", func() { + status := map[string]interface{}{ + "conditions": []interface{}{ + map[string]interface{}{ + "type": "Ready", + "status": "True", + }, + }, + } + Expect(isMlflowReady(status)).To(BeTrue()) + }) + + It("should return false when Ready condition is False", func() { + status := map[string]interface{}{ + "conditions": []interface{}{ + map[string]interface{}{ + "type": "Ready", + "status": "False", + }, + }, + } + Expect(isMlflowReady(status)).To(BeFalse()) + }) + + It("should return true when conditions exist but no Ready type", func() { + status := map[string]interface{}{ + "conditions": []interface{}{ + map[string]interface{}{ + "type": "Progressing", + "status": "True", + }, + }, + } + Expect(isMlflowReady(status)).To(BeTrue()) + }) + }) + + Context("injectMlflowEnv", func() { + It("should inject MLFLOW_TRACKING_AUTH and MLFLOW_TRACKING_URI when enabled", func() { + featureStore := minimalFeatureStore() + trackingUri := "https://mlflow.svc:8443" + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &trackingUri, + } + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + container := &corev1.Container{Name: "test"} + feast.injectMlflowEnv(container) + + Expect(container.Env).To(ContainElement(corev1.EnvVar{ + Name: "MLFLOW_TRACKING_AUTH", Value: "kubernetes-namespaced", + })) + Expect(container.Env).To(ContainElement(corev1.EnvVar{ + Name: "MLFLOW_TRACKING_URI", Value: trackingUri, + })) + }) + + It("should inject only MLFLOW_TRACKING_AUTH when trackingUri is nil", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{Enabled: true} + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + container := &corev1.Container{Name: "test"} + feast.injectMlflowEnv(container) + + Expect(container.Env).To(ContainElement(corev1.EnvVar{ + Name: "MLFLOW_TRACKING_AUTH", Value: "kubernetes-namespaced", + })) + for _, env := range container.Env { + Expect(env.Name).NotTo(Equal("MLFLOW_TRACKING_URI")) + } + }) + + It("should not inject env vars when mlflow is disabled", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = &feastdevv1.MlflowConfig{Enabled: false} + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + container := &corev1.Container{Name: "test"} + feast.injectMlflowEnv(container) + + Expect(container.Env).To(BeEmpty()) + }) + + It("should not inject env vars when mlflow is nil", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.Mlflow = nil + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + container := &corev1.Container{Name: "test"} + feast.injectMlflowEnv(container) + + Expect(container.Env).To(BeEmpty()) + }) + }) + + Context("HasMlflowCRD", func() { + It("should return false by default", func() { + Expect(HasMlflowCRD()).To(BeFalse()) + }) + + It("should return true when set", func() { + testSetHasMlflowCRD(true) + defer testSetHasMlflowCRD(false) + Expect(HasMlflowCRD()).To(BeTrue()) + }) + }) +}) diff --git a/infra/feast-operator/internal/controller/services/services.go b/infra/feast-operator/internal/controller/services/services.go index 4d5429bbbde..354b495b8a0 100644 --- a/infra/feast-operator/internal/controller/services/services.go +++ b/infra/feast-operator/internal/controller/services/services.go @@ -41,6 +41,7 @@ import ( // Apply defaults and set service hostnames in FeatureStore status func (feast *FeastServices) ApplyDefaults() error { ApplyDefaultsToStatus(feast.Handler.FeatureStore) + feast.applyMlflowDefaults() if err := feast.setTlsDefaults(); err != nil { return err } @@ -50,6 +51,35 @@ func (feast *FeastServices) ApplyDefaults() error { return nil } +// applyMlflowDefaults auto-enables MLflow integration when: +// - spec.mlflow is nil (not explicitly configured) — auto-discover from cluster MLflow CR +// - spec.mlflow.enabled is true but trackingUri is omitted — auto-discover the URI +// +// When spec.mlflow.enabled is explicitly false, the applied config is cleared (opt-out). +func (feast *FeastServices) applyMlflowDefaults() { + cr := feast.Handler.FeatureStore + if cr.Spec.Mlflow != nil { + if !cr.Spec.Mlflow.Enabled { + cr.Status.Applied.Mlflow = nil + return + } + // enabled: true but no trackingUri → discover it + if cr.Spec.Mlflow.TrackingUri == nil { + if uri, ok := DiscoverMlflowTrackingUri(feast.Handler.Context, feast.Handler.Client); ok { + cr.Status.Applied.Mlflow.TrackingUri = &uri + } + } + return + } + // spec.mlflow is nil → attempt auto-discovery + if uri, ok := DiscoverMlflowTrackingUri(feast.Handler.Context, feast.Handler.Client); ok { + cr.Status.Applied.Mlflow = &feastdevv1.MlflowConfig{ + Enabled: true, + TrackingUri: &uri, + } + } +} + // Deploy the feast services func (feast *FeastServices) Deploy() error { if feast.noLocalCoreServerConfigured() { @@ -94,6 +124,9 @@ func (feast *FeastServices) Deploy() error { if err := feast.deployClient(); err != nil { return err } + if err := feast.deployMlflowRoleBinding(); err != nil { + return err + } if err := feast.deployNamespaceRegistry(); err != nil { return err } @@ -533,10 +566,30 @@ func (feast *FeastServices) setContainer(containers *[]corev1.Container, feastTy if len(volumeMounts) > 0 { container.VolumeMounts = append(container.VolumeMounts, volumeMounts...) } + feast.injectMlflowEnv(container) *containers = append(*containers, *container) } } +// injectMlflowEnv adds MLFLOW_TRACKING_AUTH and MLFLOW_TRACKING_URI env vars +// to the container when MLflow integration is enabled. +func (feast *FeastServices) injectMlflowEnv(container *corev1.Container) { + applied := feast.Handler.FeatureStore.Status.Applied.Mlflow + if applied == nil || !applied.Enabled { + return + } + mlflowEnv := []corev1.EnvVar{ + {Name: "MLFLOW_TRACKING_AUTH", Value: "kubernetes-namespaced"}, + } + if applied.TrackingUri != nil { + mlflowEnv = append(mlflowEnv, corev1.EnvVar{ + Name: "MLFLOW_TRACKING_URI", + Value: *applied.TrackingUri, + }) + } + container.Env = envOverride(container.Env, mlflowEnv) +} + func getContainer(name, workingDir string, cmd []string, containerConfigs feastdevv1.ContainerConfigs, fsYamlB64 string) *corev1.Container { container := &corev1.Container{ Name: name, diff --git a/infra/feast-operator/internal/controller/services/services_types.go b/infra/feast-operator/internal/controller/services/services_types.go index 098362af96b..9fb7077547c 100644 --- a/infra/feast-operator/internal/controller/services/services_types.go +++ b/infra/feast-operator/internal/controller/services/services_types.go @@ -316,6 +316,7 @@ type RepoConfig struct { FeatureServer *FeatureServerYamlConfig `yaml:"feature_server,omitempty"` Materialization *MaterializationYamlConfig `yaml:"materialization,omitempty"` OpenLineage *OpenLineageYamlConfig `yaml:"openlineage,omitempty"` + Mlflow *MlflowYamlConfig `yaml:"mlflow,omitempty"` DataQualityMonitoring *DataQualityMonitoringYamlConfig `yaml:"data_quality_monitoring,omitempty"` } @@ -380,6 +381,20 @@ type OpenLineageConsumerYamlConfig struct { NamespaceMapping map[string]string `yaml:"namespace_mapping,omitempty"` } +// MlflowYamlConfig maps to the mlflow section of feature_store.yaml. +// ExtraConfig is merged inline so additional key-value pairs appear at the same +// YAML level as the typed fields. +type MlflowYamlConfig struct { + Enabled bool `yaml:"enabled"` + TrackingUri *string `yaml:"tracking_uri,omitempty"` + AutoLog *bool `yaml:"auto_log,omitempty"` + AutoLogEntityDf *bool `yaml:"auto_log_entity_df,omitempty"` + EntityDfMaxRows *int32 `yaml:"entity_df_max_rows,omitempty"` + LogOperations *bool `yaml:"log_operations,omitempty"` + OpsExperimentSuffix *string `yaml:"ops_experiment_suffix,omitempty"` + ExtraConfig map[string]interface{} `yaml:",inline,omitempty"` +} + // OfflineStoreConfig is the configuration that relates to reading from and writing to the Feast offline store. type OfflineStoreConfig struct { Host string `yaml:"host,omitempty"` diff --git a/infra/feast-operator/internal/controller/services/suite_test.go b/infra/feast-operator/internal/controller/services/suite_test.go index de1b75817ef..bdef3144d92 100644 --- a/infra/feast-operator/internal/controller/services/suite_test.go +++ b/infra/feast-operator/internal/controller/services/suite_test.go @@ -92,3 +92,7 @@ func testSetIsOpenShift() { func testSetHasServiceMonitorCRD(val bool) { hasServiceMonitorCRD = val } + +func testSetHasMlflowCRD(val bool) { + hasMlflowCRD = val +} diff --git a/infra/feast-operator/internal/controller/services/util.go b/infra/feast-operator/internal/controller/services/util.go index 387ccdb631b..89fdb291079 100644 --- a/infra/feast-operator/internal/controller/services/util.go +++ b/infra/feast-operator/internal/controller/services/util.go @@ -22,6 +22,7 @@ import ( var isOpenShift = false var hasServiceMonitorCRD = false +var hasMlflowCRD = false func IsRegistryServer(featureStore *feastdevv1.FeatureStore) bool { return IsLocalRegistry(featureStore) && featureStore.Status.Applied.Services.Registry.Local.Server != nil @@ -381,6 +382,12 @@ func HasServiceMonitorCRD() bool { return hasServiceMonitorCRD } +// HasMlflowCRD returns whether the mlflow.opendatahub.io API group +// (MLflow Operator) is available in the cluster. +func HasMlflowCRD() bool { + return hasMlflowCRD +} + // SetIsOpenShift sets the global flag isOpenShift by the controller manager. // We don't need to keep fetching the API every reconciliation cycle that we need to know about the platform. func SetIsOpenShift(cfg *rest.Config) { @@ -404,6 +411,9 @@ func SetIsOpenShift(cfg *rest.Config) { if v.Name == "monitoring.coreos.com" { hasServiceMonitorCRD = true } + if v.Name == "mlflow.opendatahub.io" { + hasMlflowCRD = true + } } }