Add MLflowEnabled config option.

Signed-off-by: Alyssa Goins <agoins@redhat.com>

Author: alyssacgoins

api/v1/dspipeline_types.go

@@ -55,6 +55,8 @@ type DSPASpec struct {
 	// Proxy configuration for all DSPA components to enable usage in environments requiring proxy access
 	// +kubebuilder:validation:Optional
 	Proxy *ProxyConfig `json:"proxy,omitempty"`
+
+	MLflow *MLflowConfig `json:"mlflowConfig,omitEmpty"`
 }
 
 // +kubebuilder:validation:Pattern=`^(Managed|Removed)$`
@@ -466,6 +468,18 @@ type ComponentDetailStatus struct {
 	ExternalUrl string `json:"externalUrl,omitempty"`
 }
 
+type IntegrationMode int
+
+const (
+	AutoDetect IntegrationMode = iota
+	Disabled
+)
+
+type MLflowConfig struct {
+	IntegrationMode
+	InjectUserEnvVars bool
+}
+
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
 //+kubebuilder:resource:shortName=dspa

config/crd/bases/datasciencepipelinesapplications.opendatahub.io_datasciencepipelinesapplications.yaml

@@ -914,6 +914,21 @@ spec:
                         type: object
                     type: object
                 type: object
+              mlflow:
+                description: |-
+                  MLflow can be optionally enabled as an API server plugin for tracking pipeline runs 
+                  in MLflow experiments.
+                properties:
+                  IntegrationMode:
+                    description: |- 
+                      Indicates whether to enable MLflow plugin on API server if plugin installed on cluster, 
+                      or disable entirely.
+                    enum:
+                      - "AUTODETECT"
+                      - "DISABLED"
+                  InjectUserEnvVars:
+                    description: Indicates if user container env variable should be updated by KFP.
+                    type: boolean
               podToPodTLS:
                 default: true
                 description: PodToPodTLS Set to "true" or "false" to enable or disable

config/internal/apiserver/default/rolebinding_ds-pipeline.yaml.tmpl

@@ -13,3 +13,21 @@ roleRef:
 subjects:
   - kind: ServiceAccount
     name: {{.APIServerDefaultResourceName}}
+
+---
+# mlflow-integration binding
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: mlflow-integration
+  namespace: {{.Namespace}}
+  labels:
+    app: {{.APIServerDefaultResourceName}}
+    component: data-science-pipelines
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: mlflow-integration
+subjects:
+  - kind: ServiceAccount
+    name: {{.APIServerDefaultResourceName}}

config/internal/apiserver/default/rolebinding_pipeline-runner.yaml.tmpl

@@ -13,3 +13,21 @@ roleRef:
 subjects:
   - kind: ServiceAccount
     name: pipeline-runner-{{.Name}}
+
+---
+# mlflow-integration binding
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: mlflow-integration
+  namespace: {{.Namespace}}
+  labels:
+    app: {{.APIServerDefaultResourceName}}
+    component: data-science-pipelines
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: mlflow-integration
+subjects:
+  - kind: ServiceAccount
+    name: pipeline-runner-{{.Name}}

controllers/dspipeline_params.go

@@ -41,6 +41,8 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	mlflowv1 "github.com/opendatahub-io/mlflow-operator/api/v1"
 )
 
 const MlmdIsRequired = "MLMD explicitly disabled in DSPA, but is a required component for DSP"
@@ -101,6 +103,7 @@ type DSPAParams struct {
 
 	APIServerServiceDNSName string
 	FIPSEnabled             bool
+	MLflow                  *dspa.MLflowConfig
 
 	// Proxy configuration for all DSPA components
 	ProxyConfig *dspa.ProxyConfig
@@ -141,6 +144,29 @@ type ObjectStorageConnection struct {
 	ExternalRouteURL  string
 }
 
+type PluginConfig struct {
+	Endpoint string               `json:"endpoint,omitempty" mapstructure:"endpoint"`
+	Timeout  string               `json:"timeout,omitempty" mapstructure:"timeout"`
+	TLS      TLSConfig            `json:"tls,omitempty" mapstructure:"tls"`
+	Settings MLflowPluginSettings `json:"settings,omitempty" mapstructure:"settings"`
+}
+
+type TLSConfig struct {
+	InsecureSkipVerify bool   `json:"insecureSkipVerify,omitempty" mapstructure:"insecureSkipVerify"`
+	CABundlePath       string `json:"caBundlePath,omitempty" mapstructure:"caBundlePath"`
+}
+
+// MLflowPluginSettings contains MLflow-specific settings to be marshalled into PluginConfig.Settings
+type MLflowPluginSettings struct {
+	WorkspacesEnabled     bool   `json:"workspacesEnabled,omitempty"`
+	ExperimentDescription string `json:"experimentDescription,omitempty"`
+	//todo: add default experiment name?
+	DefaultExperimentName string `json:"defaultExperimentName"`
+	//todo: KFP base URL?
+	KFPBaseURL        string `json:"kfpBaseURL,omitempty"`
+	InjectUserEnvVars bool   `json:"injectUserEnvVars,omitempty"`
+}
+
 // UsingExternalDB will return true if an external Database is specified in the CR, otherwise false.
 func (p *DSPAParams) UsingExternalDB(dsp *dspa.DataSciencePipelinesApplication) bool {
 	if dsp.Spec.Database != nil && dsp.Spec.Database.ExternalDB != nil {
@@ -217,6 +243,25 @@ func (p *DSPAParams) RetrieveSecret(ctx context.Context, client client.Client, s
 	return base64.StdEncoding.EncodeToString(secret.Data[secretKey]), nil
 }
 
+func (p *DSPAParams) RetrieveMLflowEndpoint(ctx context.Context, client client.Client, log logr.Logger) (string, error) {
+	mlflowName := "mlflow"
+	mlflow := &mlflowv1.MLflow{}
+	namespacedName := types.NamespacedName{
+		Name:      mlflowName,
+		Namespace: p.Namespace,
+	}
+	err := client.Get(ctx, namespacedName, mlflow)
+	if err != nil {
+		log.V(1).Info(fmt.Sprintf("Unable to retrieve mlflow resource [%s].", mlflowName))
+		return "", err
+	}
+	status := mlflow.Status
+	if status.Address != nil && status.Address.URL != "" {
+		return mlflow.Status.Address.URL, nil
+	}
+	return "", errors.New("MLflow resource missing Status.Address.URL field. Unable to resolve endpoint")
+}
+
 func (p *DSPAParams) RetrieveOrCreateSecret(ctx context.Context, client client.Client, secretName, secretKey string, generatedPasswordLength int, log logr.Logger) (string, error) {
 	val, err := p.RetrieveSecret(ctx, client, secretName, secretKey, log)
 	if err != nil && apierrs.IsNotFound(err) {
@@ -703,6 +748,23 @@ func (p *DSPAParams) ExtractParams(ctx context.Context, dsp *dspa.DataSciencePip
 		p.PodToPodTLS = *dsp.Spec.PodToPodTLS
 	}
 
+	if dsp.Spec.MLflow != nil {
+		mlflowCfg := dspa.MLflowConfig{
+			IntegrationMode:   dspa.AutoDetect,
+			InjectUserEnvVars: false,
+		}
+
+		// Override default settings if specified.
+		if dsp.Spec.MLflow.IntegrationMode == dspa.Disabled {
+			mlflowCfg.IntegrationMode = dspa.Disabled
+		}
+		if dsp.Spec.MLflow.InjectUserEnvVars == true {
+			mlflowCfg.InjectUserEnvVars = true
+		}
+
+		p.MLflow = &mlflowCfg
+	}
+
 	p.ProxyConfig = dsp.Spec.Proxy
 
 	log := loggr.WithValues("namespace", p.Namespace).WithValues("dspa_name", p.Name)
@@ -729,6 +791,27 @@ func (p *DSPAParams) ExtractParams(ctx context.Context, dsp *dspa.DataSciencePip
 
 		setResourcesDefault(config.APIServerResourceRequirements, &p.APIServer.Resources)
 
+		if p.MLflow != nil && p.MLflow.IntegrationMode == dspa.AutoDetect {
+			// Retrieve internal MLflow service endpoint for use in API Server MLflow plugin config.
+			mlflowEndpoint, err := p.RetrieveMLflowEndpoint(ctx, client, log)
+			if err == nil {
+				pluginCfg, err := BuildMLflowPluginConfigJson(mlflowEndpoint, p.CustomCABundleRootMountPath, p.MLflow.InjectUserEnvVars)
+				if err != nil {
+					//todo: do we want to return an error here? or just log an error.
+					log.Info("Failed to build MLflow plugin config. MLflow API server plugin will not be enabled.")
+					return err
+				}
+				// Append MLflow Plugin config to API server configs.
+				err = util.UpdateConfigMapByName(ctx, p.APIServer.CustomServerConfig.Name, p.Namespace, client, "config.json", pluginCfg)
+				if err != nil {
+					log.Info("Failed to update API server config with MLflow plugin. MLflow API server plugin will not be enabled.")
+					return err
+				}
+			} else {
+				log.Error(err, "failed to retrieve MLflow internal endpoint. MLflow API server plugin will not be enabled.")
+			}
+		}
+
 		if p.APIServer.CustomServerConfig == nil {
 			p.APIServer.CustomServerConfig = &dspa.ScriptConfigMap{
 				Name: config.CustomServerConfigMapNamePrefix + dsp.Name,
@@ -977,3 +1060,29 @@ func (p *DSPAParams) ExtractParams(ctx context.Context, dsp *dspa.DataSciencePip
 
 	return nil
 }
+
+func BuildMLflowPluginConfigJson(mlflowEndpoint string, caBundlePath string, injectUserEnvVars bool) (string, error) {
+	settings := MLflowPluginSettings{
+		WorkspacesEnabled:     true,
+		ExperimentDescription: "Created by AI Pipelines.",
+		DefaultExperimentName: "pipelines-exp",
+		//todo: this should potentially be passed in.
+		KFPBaseURL:        "",
+		InjectUserEnvVars: injectUserEnvVars,
+	}
+
+	pluginCfgJson, err := json.Marshal(
+		PluginConfig{
+			Endpoint: mlflowEndpoint,
+			Timeout:  "30s",
+			TLS: TLSConfig{
+				InsecureSkipVerify: false,
+				CABundlePath:       caBundlePath,
+			},
+			Settings: settings,
+		})
+	if err != nil {
+		return "", fmt.Errorf("failed to marshal MLflow plugin config: %w", err)
+	}
+	return string(pluginCfgJson), nil
+}

controllers/util/util.go

@@ -17,12 +17,14 @@ limitations under the License.
 package util
 
 import (
+	"encoding/json"
 	"fmt"
+	"os"
+	"path/filepath"
+
 	mf "github.com/manifestival/manifestival"
 	dspav1 "github.com/opendatahub-io/data-science-pipelines-operator/api/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"os"
-	"path/filepath"
 
 	"github.com/opendatahub-io/data-science-pipelines-operator/controllers/config"
 
@@ -101,6 +103,35 @@ func GetConfigMap(ctx context.Context, cfgName, ns string, client client.Client)
 	return cfgMap, nil
 }
 
+func UpdateConfigMap(ctx context.Context, cfgMap *v1.ConfigMap, client client.Client) error {
+	err := client.Update(ctx, cfgMap)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func UpdateConfigMapByName(ctx context.Context, cfgName, ns string, client client.Client, field, value string) error {
+	cm, err := GetConfigMap(ctx, cfgName, ns, client)
+	if err != nil {
+		return fmt.Errorf("failed to get configmap %s/%s: %w", ns, cfgName, err)
+	}
+	cmField := cm.Data[field]
+	if cmField == "" {
+		return fmt.Errorf("configmap field %s does not exist", field)
+	}
+	updatedField, err := updateJSON(cmField, field, value)
+	if err != nil {
+		return fmt.Errorf("failed to update configmap field %s/%s: %w", ns, field, err)
+	}
+	cm.Data[field] = updatedField
+	err = UpdateConfigMap(ctx, cm, client)
+	if err != nil {
+		return fmt.Errorf("failed to update configmap %s/%s: %w", ns, cfgName, err)
+	}
+	return nil
+}
+
 // GetConfigMapValue fetches the value for the provided configmap mapped to a given key
 func GetConfigMapValue(cfgKey string, cfgMap *v1.ConfigMap) string {
 	if val, ok := cfgMap.Data[cfgKey]; ok {
@@ -280,3 +311,17 @@ func AddDeploymentPodLabelTransformer(labelKey, labelValue string) mf.Transforme
 		return nil
 	}
 }
+
+func updateJSON(jsonStr string, field string, value any) (string, error) {
+	var obj map[string]any
+	if err := json.Unmarshal([]byte(jsonStr), &obj); err != nil {
+		return "", fmt.Errorf("invalid JSON input: %w", err)
+	}
+
+	obj[field] = value
+	out, err := json.Marshal(obj)
+	if err != nil {
+		return "", fmt.Errorf("marshal updated JSON: %w", err)
+	}
+	return string(out), nil
+}

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/manifestival/controller-runtime-client v0.4.0
 	github.com/manifestival/manifestival v0.7.2
 	github.com/minio/minio-go/v7 v7.0.99
+	github.com/opendatahub-io/mlflow-operator/api v0.0.0-20260331183147-5f8991ebee1a
 	github.com/openshift/api v0.0.0-20260331162130-f7b3bd900c75
 	github.com/prometheus/client_golang v1.23.2
 	github.com/spf13/viper v1.21.0

go.sum

@@ -1218,6 +1218,8 @@ github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1y
 github.com/onsi/gomega v1.10.2/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.38.2 h1:eZCjf2xjZAqe+LeWvKb5weQ+NcPwX84kqJ0cZNxok2A=
 github.com/onsi/gomega v1.38.2/go.mod h1:W2MJcYxRGV63b418Ai34Ud0hEdTVXq9NW9+Sx6uXf3k=
+github.com/opendatahub-io/mlflow-operator/api v0.0.0-20260331183147-5f8991ebee1a h1:mWIyP//5+UMioydwJa1RLmFt2+YnSJHpGwL+4S9xsQY=
+github.com/opendatahub-io/mlflow-operator/api v0.0.0-20260331183147-5f8991ebee1a/go.mod h1:4cBD//CLye8UPT2d2gDLrwLOVGiOv7FtAYNvMYnzLj0=
 github.com/openshift/api v0.0.0-20260331162130-f7b3bd900c75 h1:BcNL7bI9rxRQdDVsbtW0kPw23I6GtNwvBYg9rjoWenE=
 github.com/openshift/api v0.0.0-20260331162130-f7b3bd900c75/go.mod h1:pyVjK0nZ4sRs4fuQVQ4rubsJdahI1PB94LnQ8sGdvxo=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=

main.go

@@ -56,6 +56,7 @@ import (
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 	//+kubebuilder:scaffold:imports
+	mlflowv1 "github.com/opendatahub-io/mlflow-operator/api/v1"
 )
 
 var (
@@ -247,6 +248,9 @@ func main() {
 				// informer still detects changes to those external resources.
 				&corev1.ConfigMap{}: {Transform: stripConfigMapData},
 				&corev1.Secret{}:    {Transform: stripSecretData},
+				//todo: possible to add cache ttl here?
+				//todo: can we potentially add a specific sync period for this one watched resource?
+				&mlflowv1.MLflow{}: {},
 			},
 		},
 		// ConfigMap and Secret client reads bypass the cache entirely for

scripts/release/params.py

@@ -46,7 +46,8 @@
     "MANAGEDPIPELINES": "\"{}\"",
     "PLATFORMVERSION": "\"v0.0.0\"",
     "FIPSENABLED": "false",
-    "WEBHOOK_ANNOTATIONS": ""
+    "WEBHOOK_ANNOTATIONS": "",
+    "MLFLOW": "{\"}\"",
 }