From 8f9cfa243ea00cb0bfad5fbe14a639d44f256ecb Mon Sep 17 00:00:00 2001 From: Bobbins228 Date: Tue, 26 May 2026 16:00:09 +0100 Subject: [PATCH 1/2] feat: add MLflow operand controller for CR lifecycle and OTEL RBAC Adds MLflowOperandReconciler that watches DataScienceCluster and: - Creates the MLflow CR in redhat-ods-applications when DSC mlflowoperator is Managed - Waits for MLflow Service/Endpoints readiness via requeue - Creates per-agent-namespace RoleBindings granting otel-collector SA access to MLflow ClusterRoles for trace export - Re-reconciles when DSC MLflow component state changes RBAC updates: - Expand mlflows verbs to include create/update - Add datascienceclusters get/list/watch - Add clusterroles get and bind (scoped to mlflow-operator roles) - Add endpoints to core resource watches Registered under --enable-mlflow gate alongside existing MLflowReconciler. Uses DSC v2 API (mlflowoperator field is v2-only). Ref: RHAIENG-4902 Assisted-By: cursor Signed-off-by: Bobbins228 --- .../kagenti-operator/templates/rbac/role.yaml | 33 ++ kagenti-operator/cmd/main.go | 12 + .../internal/controller/mlflow_controller.go | 2 +- .../controller/mlflow_operand_controller.go | 432 ++++++++++++++ .../mlflow_operand_controller_test.go | 535 ++++++++++++++++++ kagenti-operator/internal/mlflow/types.go | 138 ++++- 6 files changed, 1148 insertions(+), 4 deletions(-) create mode 100644 kagenti-operator/internal/controller/mlflow_operand_controller.go create mode 100644 kagenti-operator/internal/controller/mlflow_operand_controller_test.go diff --git a/charts/kagenti-operator/templates/rbac/role.yaml b/charts/kagenti-operator/templates/rbac/role.yaml index c4445640..dde766fe 100755 --- a/charts/kagenti-operator/templates/rbac/role.yaml +++ b/charts/kagenti-operator/templates/rbac/role.yaml @@ -142,13 +142,31 @@ rules: - statefulsets/finalizers verbs: - update +- apiGroups: + - discovery.k8s.io + resources: + - endpointslices + verbs: + - get + - list + - watch +- apiGroups: + - datasciencecluster.opendatahub.io + resources: + - datascienceclusters + verbs: + - get + - list + - watch - apiGroups: - mlflow.opendatahub.io resources: - mlflows verbs: + - create - get - list + - update - watch - apiGroups: - mlflow.opendatahub.io @@ -201,6 +219,21 @@ rules: - list - patch - watch +- apiGroups: + - rbac.authorization.k8s.io + resources: + - clusterroles + verbs: + - get +- apiGroups: + - rbac.authorization.k8s.io + resources: + - clusterroles + resourceNames: + - mlflow-operator-mlflow-integration + - mlflow-operator-mlflow-edit + verbs: + - bind - apiGroups: - rbac.authorization.k8s.io resources: diff --git a/kagenti-operator/cmd/main.go b/kagenti-operator/cmd/main.go index 4a321e6d..46cb8e8a 100644 --- a/kagenti-operator/cmd/main.go +++ b/kagenti-operator/cmd/main.go @@ -526,6 +526,18 @@ func main() { os.Exit(1) } setupLog.Info("MLflow UI config bootstrap enabled") + + if err = (&controller.MLflowOperandReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + //nolint:staticcheck // consistent with existing controllers + Recorder: mgr.GetEventRecorderFor("mlflow-operand-controller"), + OperatorNamespace: getOperatorNamespace(), + }).SetupWithManager(mgr); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "MLflowOperand") + os.Exit(1) + } + setupLog.Info("MLflow operand controller enabled") } if enableClientRegistration { diff --git a/kagenti-operator/internal/controller/mlflow_controller.go b/kagenti-operator/internal/controller/mlflow_controller.go index 007cf18b..4e1333bb 100644 --- a/kagenti-operator/internal/controller/mlflow_controller.go +++ b/kagenti-operator/internal/controller/mlflow_controller.go @@ -76,7 +76,7 @@ type MLflowReconciler struct { ResolveTrackingURI func(ctx context.Context) string } -// +kubebuilder:rbac:groups=mlflow.opendatahub.io,resources=mlflows,verbs=get;list;watch +// +kubebuilder:rbac:groups=mlflow.opendatahub.io,resources=mlflows,verbs=create;get;list;update;watch // +kubebuilder:rbac:groups=mlflow.opendatahub.io,resources=mlflowexperiments,verbs=get;list;watch;update // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=roles,verbs=create;get;list;watch;update // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=rolebindings,verbs=create;get;list;watch;update diff --git a/kagenti-operator/internal/controller/mlflow_operand_controller.go b/kagenti-operator/internal/controller/mlflow_operand_controller.go new file mode 100644 index 00000000..38bf4f7a --- /dev/null +++ b/kagenti-operator/internal/controller/mlflow_operand_controller.go @@ -0,0 +1,432 @@ +/* +Copyright 2026. + +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 controller + +import ( + "context" + "fmt" + "strings" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/kagenti/operator/internal/mlflow" +) + +const ( + // MLflowOperandNamespace is the namespace where RHOAI MLflow CRs live. + MLflowOperandNamespace = "redhat-ods-applications" + + // MLflowOperandName is the default name for the operator-managed MLflow CR. + MLflowOperandName = "mlflow" + + // OTELCollectorRoleBindingName is the name used for the per-namespace + // RoleBinding granting the otel-collector SA access to MLflow. + OTELCollectorRoleBindingName = "otel-collector-mlflow" + + // OTELCollectorSAName is the ServiceAccount used by the OTEL collector. + OTELCollectorSAName = "otel-collector" + + // DSCName is the well-known name of the singleton DataScienceCluster. + DSCName = "default-dsc" + + // MLflowClusterRoleIntegration is the ClusterRole created by the MLflow operator. + MLflowClusterRoleIntegration = "mlflow-operator-mlflow-integration" + + // MLflowClusterRoleEdit is the newer ClusterRole name (preferred). + MLflowClusterRoleEdit = "mlflow-operator-mlflow-edit" + + // requeueMLflowNotReady is the requeue delay when MLflow is not yet available. + requeueMLflowNotReady = 30 * time.Second +) + +// MLflowOperandReconciler watches DataScienceCluster resources and ensures: +// 1. An MLflow CR exists in redhat-ods-applications when the DSC has mlflowoperator: Managed +// 2. Per-agent-namespace RoleBindings exist for the otel-collector SA +// +// It is registered under the same --enable-mlflow gate as MLflowReconciler. +type MLflowOperandReconciler struct { + client.Client + Scheme *runtime.Scheme + Recorder record.EventRecorder + + // OperatorNamespace is the namespace the operator runs in (default: kagenti-system). + OperatorNamespace string + + // ResolveMLflowClusterRole overrides ClusterRole discovery for testing. + ResolveMLflowClusterRole func(ctx context.Context) string +} + +// +kubebuilder:rbac:groups=datasciencecluster.opendatahub.io,resources=datascienceclusters,verbs=get;list;watch +// +kubebuilder:rbac:groups=mlflow.opendatahub.io,resources=mlflows,verbs=create;get;list;update;watch +// +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=rolebindings,verbs=create;delete;get;list;watch;update +// +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterroles,verbs=get +// +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterroles,resourceNames=mlflow-operator-mlflow-integration;mlflow-operator-mlflow-edit,verbs=bind +// +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch +// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch + +func (r *MLflowOperandReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + logger := log.FromContext(ctx) + logger.V(1).Info("Reconciling MLflow operand", "name", req.Name) + + dsc := &mlflow.DataScienceCluster{} + if err := r.Get(ctx, req.NamespacedName, dsc); err != nil { + if errors.IsNotFound(err) { + logger.V(1).Info("DataScienceCluster not found, nothing to do") + return ctrl.Result{}, nil + } + return ctrl.Result{}, err + } + + if dsc.Spec.Components.MLflowOperator.ManagementState != "Managed" { + logger.V(1).Info("DSC mlflowoperator is not Managed, skipping", + "state", dsc.Spec.Components.MLflowOperator.ManagementState) + return ctrl.Result{}, nil + } + + if result, err := r.ensureMLflowCR(ctx); err != nil || result.RequeueAfter > 0 { + return result, err + } + + if result, err := r.waitForMLflowReady(ctx); err != nil || result.RequeueAfter > 0 { + return result, err + } + + if err := r.ensureOTELRoleBindings(ctx); err != nil { + return ctrl.Result{}, err + } + + logger.Info("MLflow operand reconciliation complete") + return ctrl.Result{}, nil +} + +// ensureMLflowCR creates the MLflow CR in redhat-ods-applications if absent. +func (r *MLflowOperandReconciler) ensureMLflowCR(ctx context.Context) (ctrl.Result, error) { + logger := log.FromContext(ctx) + + existing := &mlflow.MLflow{} + err := r.Get(ctx, types.NamespacedName{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, existing) + + if err == nil { + logger.V(1).Info("MLflow CR already exists", "name", MLflowOperandName, "namespace", MLflowOperandNamespace) + return ctrl.Result{}, nil + } + if !errors.IsNotFound(err) { + return ctrl.Result{}, fmt.Errorf("checking MLflow CR: %w", err) + } + + cr := &mlflow.MLflow{ + TypeMeta: metav1.TypeMeta{ + APIVersion: mlflow.SchemeGroupVersion.String(), + Kind: "MLflow", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + Labels: map[string]string{ + LabelManagedBy: LabelManagedByValue, + }, + }, + Spec: mlflow.MLflowSpec{ + Storage: &mlflow.MLflowStorage{ + AccessModes: []string{"ReadWriteOnce"}, + Resources: &mlflow.MLflowStorageResourceReqs{ + Requests: map[string]string{ + "storage": "10Gi", + }, + }, + }, + BackendStoreURI: "sqlite:////mlflow/mlflow.db", + ArtifactsDestination: "file:///mlflow/artifacts", + ServeArtifacts: true, + }, + } + + if err := r.Create(ctx, cr); err != nil { + if errors.IsAlreadyExists(err) { + logger.V(1).Info("MLflow CR was created concurrently") + return ctrl.Result{}, nil + } + return ctrl.Result{}, fmt.Errorf("creating MLflow CR: %w", err) + } + + logger.Info("Created MLflow CR", "name", MLflowOperandName, "namespace", MLflowOperandNamespace) + if r.Recorder != nil { + r.Recorder.Event(cr, "Normal", "MLflowCRCreated", + fmt.Sprintf("Created MLflow CR %s/%s", MLflowOperandNamespace, MLflowOperandName)) + } + + return ctrl.Result{RequeueAfter: requeueMLflowNotReady}, nil +} + +// waitForMLflowReady checks that the MLflow Service has at least one ready endpoint. +func (r *MLflowOperandReconciler) waitForMLflowReady(ctx context.Context) (ctrl.Result, error) { + logger := log.FromContext(ctx) + + svc := &corev1.Service{} + err := r.Get(ctx, types.NamespacedName{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, svc) + if err != nil { + if errors.IsNotFound(err) { + logger.V(1).Info("MLflow Service not yet created, requeueing") + return ctrl.Result{RequeueAfter: requeueMLflowNotReady}, nil + } + return ctrl.Result{}, fmt.Errorf("checking MLflow Service: %w", err) + } + + epSlices := &discoveryv1.EndpointSliceList{} + err = r.List(ctx, epSlices, + client.InNamespace(MLflowOperandNamespace), + client.MatchingLabels{"kubernetes.io/service-name": MLflowOperandName}, + ) + if err != nil { + return ctrl.Result{}, fmt.Errorf("listing MLflow EndpointSlices: %w", err) + } + + for i := range epSlices.Items { + for j := range epSlices.Items[i].Endpoints { + if epSlices.Items[i].Endpoints[j].Conditions.Ready != nil && + *epSlices.Items[i].Endpoints[j].Conditions.Ready { + logger.V(1).Info("MLflow Service has ready endpoints") + return ctrl.Result{}, nil + } + } + } + + logger.V(1).Info("MLflow Service has no ready endpoints, requeueing") + return ctrl.Result{RequeueAfter: requeueMLflowNotReady}, nil +} + +// ensureOTELRoleBindings creates a RoleBinding in every namespace that contains +// an agent Deployment, granting the otel-collector SA access to the MLflow +// ClusterRole for trace export. +func (r *MLflowOperandReconciler) ensureOTELRoleBindings(ctx context.Context) error { + logger := log.FromContext(ctx) + + clusterRole := r.resolveMLflowClusterRole(ctx) + if clusterRole == "" { + logger.Info("No MLflow ClusterRole found, skipping OTEL RoleBindings") + return nil + } + + namespaces, err := r.agentNamespaces(ctx) + if err != nil { + return fmt.Errorf("listing agent namespaces: %w", err) + } + + operatorNS := r.operatorNamespace() + + // Also ensure RoleBinding in the MLflow namespace itself. + allNamespaces := append(namespaces, MLflowOperandNamespace) + + for _, ns := range allNamespaces { + if err := r.ensureOTELRoleBinding(ctx, ns, operatorNS, clusterRole); err != nil { + return err + } + } + + logger.V(1).Info("OTEL RoleBindings ensured", "namespaces", allNamespaces, "clusterRole", clusterRole) + return nil +} + +func (r *MLflowOperandReconciler) ensureOTELRoleBinding(ctx context.Context, namespace, operatorNS, clusterRole string) error { + desired := rbacv1.RoleRef{ + APIGroup: rbacv1.GroupName, + Kind: "ClusterRole", + Name: clusterRole, + } + + rb := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: OTELCollectorRoleBindingName, + Namespace: namespace, + }, + } + + _, err := controllerutil.CreateOrUpdate(ctx, r.Client, rb, func() error { + rb.Labels = map[string]string{ + LabelManagedBy: LabelManagedByValue, + } + rb.RoleRef = desired + rb.Subjects = []rbacv1.Subject{ + { + Kind: rbacv1.ServiceAccountKind, + Name: OTELCollectorSAName, + Namespace: operatorNS, + }, + } + return nil + }) + + if err != nil && strings.Contains(err.Error(), "cannot change roleRef") { + logger := log.FromContext(ctx) + logger.Info("RoleBinding roleRef mismatch, recreating", + "namespace", namespace, "desiredClusterRole", clusterRole) + if delErr := r.Delete(ctx, rb); delErr != nil { + return fmt.Errorf("deleting stale OTEL RoleBinding in %s: %w", namespace, delErr) + } + rb = &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: OTELCollectorRoleBindingName, + Namespace: namespace, + Labels: map[string]string{LabelManagedBy: LabelManagedByValue}, + }, + RoleRef: desired, + Subjects: []rbacv1.Subject{ + { + Kind: rbacv1.ServiceAccountKind, + Name: OTELCollectorSAName, + Namespace: operatorNS, + }, + }, + } + if createErr := r.Create(ctx, rb); createErr != nil { + return fmt.Errorf("recreating OTEL RoleBinding in %s: %w", namespace, createErr) + } + return nil + } + if err != nil { + return fmt.Errorf("ensuring OTEL RoleBinding in %s: %w", namespace, err) + } + return nil +} + +// resolveMLflowClusterRole finds the MLflow ClusterRole, preferring the newer +// "edit" name, falling back to "integration". +func (r *MLflowOperandReconciler) resolveMLflowClusterRole(ctx context.Context) string { + if r.ResolveMLflowClusterRole != nil { + return r.ResolveMLflowClusterRole(ctx) + } + + for _, name := range []string{MLflowClusterRoleEdit, MLflowClusterRoleIntegration} { + cr := &rbacv1.ClusterRole{} + if err := r.Get(ctx, types.NamespacedName{Name: name}, cr); err == nil { + return name + } + } + return "" +} + +// agentNamespaces returns the set of namespaces containing at least one +// Deployment with the kagenti.io/type=agent label. +func (r *MLflowOperandReconciler) agentNamespaces(ctx context.Context) ([]string, error) { + depList := &appsv1.DeploymentList{} + if err := r.List(ctx, depList, client.MatchingLabels{ + LabelAgentType: LabelValueAgent, + }); err != nil { + return nil, err + } + + seen := make(map[string]struct{}) + var namespaces []string + for i := range depList.Items { + ns := depList.Items[i].Namespace + if _, ok := seen[ns]; !ok { + seen[ns] = struct{}{} + namespaces = append(namespaces, ns) + } + } + return namespaces, nil +} + +func (r *MLflowOperandReconciler) operatorNamespace() string { + if r.OperatorNamespace != "" { + return r.OperatorNamespace + } + return "kagenti-system" +} + +// dscMLflowChangePredicate fires only when the DSC mlflowoperator managementState changes. +func dscMLflowChangePredicate() predicate.Predicate { + return predicate.Funcs{ + UpdateFunc: func(e event.UpdateEvent) bool { + oldDSC, ok1 := e.ObjectOld.(*mlflow.DataScienceCluster) + newDSC, ok2 := e.ObjectNew.(*mlflow.DataScienceCluster) + if !ok1 || !ok2 { + return true + } + return oldDSC.Spec.Components.MLflowOperator.ManagementState != + newDSC.Spec.Components.MLflowOperator.ManagementState + }, + CreateFunc: func(_ event.CreateEvent) bool { return true }, + DeleteFunc: func(_ event.DeleteEvent) bool { return true }, + GenericFunc: func(_ event.GenericEvent) bool { return true }, + } +} + +// SetupWithManager registers the MLflow operand controller. +// It watches DataScienceCluster as the primary resource and enqueues +// reconciliation when agent Deployments appear (to create OTEL RoleBindings +// in new namespaces). +func (r *MLflowOperandReconciler) SetupWithManager(mgr ctrl.Manager) error { + // Check if the DataScienceCluster CRD is registered. If RHOAI is not + // installed, this controller has nothing to reconcile. + _, err := mgr.GetRESTMapper().RESTMapping( + mlflow.DSCSchemeGroupVersion.WithKind("DataScienceCluster").GroupKind(), + mlflow.DSCSchemeGroupVersion.Version, + ) + if err != nil { + if meta.IsNoMatchError(err) { + log.Log.Info("DataScienceCluster CRD not registered, MLflow operand controller will not start") + return nil + } + return fmt.Errorf("checking DataScienceCluster CRD: %w", err) + } + + return ctrl.NewControllerManagedBy(mgr). + For(&mlflow.DataScienceCluster{}, builder.WithPredicates(dscMLflowChangePredicate())). + // Re-reconcile when new agent Deployments appear so we can create + // OTEL RoleBindings in their namespaces. + Watches(&appsv1.Deployment{}, + handler.EnqueueRequestsFromMapFunc(r.mapAgentDeploymentToDSC), + builder.WithPredicates(agentLabelPredicate()), + ). + Named("mlflow-operand"). + Complete(r) +} + +// mapAgentDeploymentToDSC maps any agent Deployment event to a reconcile +// request for the singleton DataScienceCluster so the operand controller +// can evaluate whether OTEL RoleBindings need creating. +func (r *MLflowOperandReconciler) mapAgentDeploymentToDSC(_ context.Context, _ client.Object) []reconcile.Request { + return []reconcile.Request{ + {NamespacedName: types.NamespacedName{Name: DSCName}}, + } +} diff --git a/kagenti-operator/internal/controller/mlflow_operand_controller_test.go b/kagenti-operator/internal/controller/mlflow_operand_controller_test.go new file mode 100644 index 00000000..30d378ab --- /dev/null +++ b/kagenti-operator/internal/controller/mlflow_operand_controller_test.go @@ -0,0 +1,535 @@ +/* +Copyright 2026. + +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 controller + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/kagenti/operator/internal/mlflow" +) + +func newMLflowTestScheme() *runtime.Scheme { + s := runtime.NewScheme() + _ = corev1.AddToScheme(s) + _ = discoveryv1.AddToScheme(s) + _ = appsv1.AddToScheme(s) + _ = rbacv1.AddToScheme(s) + _ = mlflow.AddToScheme(s) + return s +} + +func readyEndpointSlice() *discoveryv1.EndpointSlice { + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName + "-abc", + Namespace: MLflowOperandNamespace, + Labels: map[string]string{"kubernetes.io/service-name": MLflowOperandName}, + }, + Endpoints: []discoveryv1.Endpoint{ + {Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)}}, + }, + } +} + +func notReadyEndpointSlice() *discoveryv1.EndpointSlice { + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName + "-abc", + Namespace: MLflowOperandNamespace, + Labels: map[string]string{"kubernetes.io/service-name": MLflowOperandName}, + }, + Endpoints: []discoveryv1.Endpoint{ + {Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(false)}}, + }, + } +} + +func newDSC(managementState string) *mlflow.DataScienceCluster { + return &mlflow.DataScienceCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: DSCName, + }, + Spec: mlflow.DSCSpec{ + Components: mlflow.DSCComponents{ + MLflowOperator: mlflow.DSCComponentState{ + ManagementState: managementState, + }, + }, + }, + } +} + +func newOperandReconciler(cl client.Client, scheme *runtime.Scheme) *MLflowOperandReconciler { + return &MLflowOperandReconciler{ + Client: cl, + Scheme: scheme, + OperatorNamespace: "kagenti-system", + ResolveMLflowClusterRole: func(_ context.Context) string { + return MLflowClusterRoleIntegration + }, + } +} + +var _ = Describe("MLflow Operand Controller", func() { + ctx := context.Background() + + reconcileRequest := func() reconcile.Request { + return reconcile.Request{ + NamespacedName: types.NamespacedName{Name: DSCName}, + } + } + + Context("When DataScienceCluster does not exist", func() { + It("should return without error (RHOAI absent)", func() { + s := newMLflowTestScheme() + cl := fake.NewClientBuilder().WithScheme(s).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + }) + }) + + Context("When DSC mlflowoperator is not Managed", func() { + It("should skip when managementState is Removed", func() { + s := newMLflowTestScheme() + dsc := newDSC("Removed") + cl := fake.NewClientBuilder().WithScheme(s).WithObjects(dsc).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + // MLflow CR should NOT be created + mlflowCR := &mlflow.MLflow{} + err = cl.Get(ctx, types.NamespacedName{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, mlflowCR) + Expect(err).To(HaveOccurred()) + }) + + It("should skip when managementState is empty", func() { + s := newMLflowTestScheme() + dsc := newDSC("") + cl := fake.NewClientBuilder().WithScheme(s).WithObjects(dsc).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + }) + }) + + Context("When DSC mlflowoperator is Managed", func() { + It("should create the MLflow CR and requeue", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + // The namespace must exist for the fake client. + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + cl := fake.NewClientBuilder().WithScheme(s).WithObjects(dsc, ns).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(Equal(requeueMLflowNotReady)) + + // Verify MLflow CR was created with correct spec + mlflowCR := &mlflow.MLflow{} + Expect(cl.Get(ctx, types.NamespacedName{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, mlflowCR)).To(Succeed()) + + Expect(mlflowCR.Labels[LabelManagedBy]).To(Equal(LabelManagedByValue)) + Expect(mlflowCR.Spec.BackendStoreURI).To(Equal("sqlite:////mlflow/mlflow.db")) + Expect(mlflowCR.Spec.ArtifactsDestination).To(Equal("file:///mlflow/artifacts")) + Expect(mlflowCR.Spec.ServeArtifacts).To(BeTrue()) + Expect(mlflowCR.Spec.Storage).NotTo(BeNil()) + Expect(mlflowCR.Spec.Storage.AccessModes).To(Equal([]string{"ReadWriteOnce"})) + Expect(mlflowCR.Spec.Storage.Resources.Requests["storage"]).To(Equal("10Gi")) + }) + + It("should skip creation when MLflow CR already exists", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, + } + cl := fake.NewClientBuilder().WithScheme(s).WithObjects(dsc, ns, existingMLflow).Build() + r := newOperandReconciler(cl, s) + + // MLflow CR exists but Service does not — should requeue waiting for readiness + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(Equal(requeueMLflowNotReady)) + }) + + It("should wait when MLflow Service has no ready endpoints", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, + } + cl := fake.NewClientBuilder().WithScheme(s).WithObjects(dsc, ns, existingMLflow, svc, notReadyEndpointSlice()).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(Equal(requeueMLflowNotReady)) + }) + + It("should proceed to OTEL RoleBindings when MLflow is ready", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + agentNS := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team1"}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, + Namespace: MLflowOperandNamespace, + }, + } + agentDep := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-agent", + Namespace: "team1", + Labels: map[string]string{ + LabelAgentType: LabelValueAgent, + }, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app": "my-agent"}, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"app": "my-agent"}, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "agent", Image: "img:latest"}}, + }, + }, + }, + } + + cl := fake.NewClientBuilder().WithScheme(s). + WithObjects(dsc, ns, agentNS, existingMLflow, svc, readyEndpointSlice(), agentDep).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + // Verify OTEL RoleBinding in agent namespace + rb := &rbacv1.RoleBinding{} + Expect(cl.Get(ctx, types.NamespacedName{ + Name: OTELCollectorRoleBindingName, Namespace: "team1", + }, rb)).To(Succeed()) + Expect(rb.RoleRef.Kind).To(Equal("ClusterRole")) + Expect(rb.RoleRef.Name).To(Equal(MLflowClusterRoleIntegration)) + Expect(rb.Subjects).To(HaveLen(1)) + Expect(rb.Subjects[0].Kind).To(Equal("ServiceAccount")) + Expect(rb.Subjects[0].Name).To(Equal(OTELCollectorSAName)) + Expect(rb.Subjects[0].Namespace).To(Equal("kagenti-system")) + Expect(rb.Labels[LabelManagedBy]).To(Equal(LabelManagedByValue)) + + // Verify OTEL RoleBinding in MLflow namespace + rb2 := &rbacv1.RoleBinding{} + Expect(cl.Get(ctx, types.NamespacedName{ + Name: OTELCollectorRoleBindingName, Namespace: MLflowOperandNamespace, + }, rb2)).To(Succeed()) + Expect(rb2.RoleRef.Name).To(Equal(MLflowClusterRoleIntegration)) + }) + + It("should create OTEL RoleBindings in multiple agent namespaces", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + ns1 := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team1"}} + ns2 := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team2"}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + dep1 := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "agent-1", Namespace: "team1", + Labels: map[string]string{LabelAgentType: LabelValueAgent}, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a1"}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "a1"}}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "c", Image: "i:l"}}}, + }, + }, + } + dep2 := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "agent-2", Namespace: "team2", + Labels: map[string]string{LabelAgentType: LabelValueAgent}, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a2"}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "a2"}}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "c", Image: "i:l"}}}, + }, + }, + } + + cl := fake.NewClientBuilder().WithScheme(s). + WithObjects(dsc, ns, ns1, ns2, existingMLflow, svc, readyEndpointSlice(), dep1, dep2).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + for _, namespace := range []string{"team1", "team2", MLflowOperandNamespace} { + rb := &rbacv1.RoleBinding{} + Expect(cl.Get(ctx, types.NamespacedName{ + Name: OTELCollectorRoleBindingName, Namespace: namespace, + }, rb)).To(Succeed(), "expected RoleBinding in namespace %s", namespace) + } + }) + + It("should skip OTEL RoleBindings when no ClusterRole exists", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + cl := fake.NewClientBuilder().WithScheme(s). + WithObjects(dsc, ns, existingMLflow, svc, readyEndpointSlice()).Build() + r := &MLflowOperandReconciler{ + Client: cl, + Scheme: s, + OperatorNamespace: "kagenti-system", + ResolveMLflowClusterRole: func(_ context.Context) string { + return "" // no ClusterRole found + }, + } + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + }) + + It("should recreate RoleBinding when roleRef changes (upgrade path)", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + agentNS := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team1"}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + agentDep := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-agent", Namespace: "team1", + Labels: map[string]string{LabelAgentType: LabelValueAgent}, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a"}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "a"}}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "c", Image: "i:l"}}}, + }, + }, + } + staleRB := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: OTELCollectorRoleBindingName, + Namespace: "team1", + Labels: map[string]string{LabelManagedBy: LabelManagedByValue}, + }, + RoleRef: rbacv1.RoleRef{ + APIGroup: rbacv1.GroupName, + Kind: "ClusterRole", + Name: MLflowClusterRoleEdit, + }, + Subjects: []rbacv1.Subject{ + {Kind: rbacv1.ServiceAccountKind, Name: OTELCollectorSAName, Namespace: "kagenti-system"}, + }, + } + + cl := fake.NewClientBuilder().WithScheme(s). + WithObjects(dsc, ns, agentNS, existingMLflow, svc, readyEndpointSlice(), agentDep, staleRB).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + rb := &rbacv1.RoleBinding{} + Expect(cl.Get(ctx, types.NamespacedName{ + Name: OTELCollectorRoleBindingName, Namespace: "team1", + }, rb)).To(Succeed()) + Expect(rb.RoleRef.Name).To(Equal(MLflowClusterRoleIntegration)) + }) + + It("should be idempotent on repeated reconciliation", func() { + s := newMLflowTestScheme() + dsc := newDSC("Managed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + agentNS := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team1"}} + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + agentDep := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-agent", Namespace: "team1", + Labels: map[string]string{LabelAgentType: LabelValueAgent}, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a"}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "a"}}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "c", Image: "i:l"}}}, + }, + }, + } + + cl := fake.NewClientBuilder().WithScheme(s). + WithObjects(dsc, ns, agentNS, existingMLflow, svc, readyEndpointSlice(), agentDep).Build() + r := newOperandReconciler(cl, s) + + // First reconcile + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + // Second reconcile — should succeed identically + result, err = r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + // RoleBinding should still exist and be correct + rb := &rbacv1.RoleBinding{} + Expect(cl.Get(ctx, types.NamespacedName{ + Name: OTELCollectorRoleBindingName, Namespace: "team1", + }, rb)).To(Succeed()) + Expect(rb.RoleRef.Name).To(Equal(MLflowClusterRoleIntegration)) + }) + }) + + Context("DSC state change from Managed to Removed", func() { + It("should be a no-op when state changes to Removed (existing resources stay)", func() { + s := newMLflowTestScheme() + dsc := newDSC("Removed") + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: MLflowOperandNamespace}} + // Existing MLflow CR and RoleBinding from a previous reconcile + existingMLflow := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, + } + existingRB := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: OTELCollectorRoleBindingName, Namespace: "team1", + }, + RoleRef: rbacv1.RoleRef{ + APIGroup: rbacv1.GroupName, + Kind: "ClusterRole", + Name: MLflowClusterRoleIntegration, + }, + } + + cl := fake.NewClientBuilder().WithScheme(s). + WithObjects(dsc, ns, existingMLflow, existingRB).Build() + r := newOperandReconciler(cl, s) + + result, err := r.Reconcile(ctx, reconcileRequest()) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(ctrl.Result{})) + + // MLflow CR should still exist (we don't delete on state change) + Expect(cl.Get(ctx, types.NamespacedName{ + Name: MLflowOperandName, Namespace: MLflowOperandNamespace, + }, &mlflow.MLflow{})).To(Succeed()) + }) + }) +}) diff --git a/kagenti-operator/internal/mlflow/types.go b/kagenti-operator/internal/mlflow/types.go index 9abb5df5..420af59e 100644 --- a/kagenti-operator/internal/mlflow/types.go +++ b/kagenti-operator/internal/mlflow/types.go @@ -26,10 +26,13 @@ var ( // SchemeGroupVersion is the GVK for the mlflows.mlflow.opendatahub.io CRD. SchemeGroupVersion = schema.GroupVersion{Group: "mlflow.opendatahub.io", Version: "v1"} - // SchemeBuilder is used to add the MLflow types to a scheme. + // DSCSchemeGroupVersion is the GVK for datascienceclusters.datasciencecluster.opendatahub.io. + DSCSchemeGroupVersion = schema.GroupVersion{Group: "datasciencecluster.opendatahub.io", Version: "v2"} + + // SchemeBuilder is used to add the MLflow and DSC types to a scheme. SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes) - // AddToScheme registers the MLflow types with a runtime.Scheme. + // AddToScheme registers the MLflow and DSC types with a runtime.Scheme. AddToScheme = SchemeBuilder.AddToScheme ) @@ -39,17 +42,41 @@ func addKnownTypes(scheme *runtime.Scheme) error { &MLflowList{}, ) metav1.AddToGroupVersion(scheme, SchemeGroupVersion) + + scheme.AddKnownTypes(DSCSchemeGroupVersion, + &DataScienceCluster{}, + &DataScienceClusterList{}, + ) + metav1.AddToGroupVersion(scheme, DSCSchemeGroupVersion) return nil } // MLflow is a minimal representation of the mlflows.mlflow.opendatahub.io/v1 CR. -// Only the fields needed for tracking URI discovery are included. type MLflow struct { metav1.TypeMeta `json:",inline"` metav1.ObjectMeta `json:"metadata,omitempty"` + Spec MLflowSpec `json:"spec,omitempty"` Status MLflowStatus `json:"status,omitempty"` } +// MLflowSpec contains the desired state for an MLflow instance. +// Only the fields used by the operator for CR creation are included. +type MLflowSpec struct { + Storage *MLflowStorage `json:"storage,omitempty"` + BackendStoreURI string `json:"backendStoreUri,omitempty"` + ArtifactsDestination string `json:"artifactsDestination,omitempty"` + ServeArtifacts bool `json:"serveArtifacts,omitempty"` +} + +type MLflowStorage struct { + AccessModes []string `json:"accessModes,omitempty"` + Resources *MLflowStorageResourceReqs `json:"resources,omitempty"` +} + +type MLflowStorageResourceReqs struct { + Requests map[string]string `json:"requests,omitempty"` +} + type MLflowStatus struct { Conditions []metav1.Condition `json:"conditions,omitempty"` // URL is the external gateway URL for the MLflow server (e.g. via the RHOAI data-science gateway). @@ -86,9 +113,40 @@ func (in *MLflow) DeepCopyInto(out *MLflow) { *out = *in out.TypeMeta = in.TypeMeta in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) in.Status.DeepCopyInto(&out.Status) } +func (in *MLflowSpec) DeepCopyInto(out *MLflowSpec) { + *out = *in + if in.Storage != nil { + out.Storage = new(MLflowStorage) + in.Storage.DeepCopyInto(out.Storage) + } +} + +func (in *MLflowStorage) DeepCopyInto(out *MLflowStorage) { + *out = *in + if in.AccessModes != nil { + out.AccessModes = make([]string, len(in.AccessModes)) + copy(out.AccessModes, in.AccessModes) + } + if in.Resources != nil { + out.Resources = new(MLflowStorageResourceReqs) + in.Resources.DeepCopyInto(out.Resources) + } +} + +func (in *MLflowStorageResourceReqs) DeepCopyInto(out *MLflowStorageResourceReqs) { + *out = *in + if in.Requests != nil { + out.Requests = make(map[string]string, len(in.Requests)) + for k, v := range in.Requests { + out.Requests[k] = v + } + } +} + func (in *MLflowStatus) DeepCopyInto(out *MLflowStatus) { *out = *in if in.Conditions != nil { @@ -126,3 +184,77 @@ func (in *MLflowList) DeepCopyInto(out *MLflowList) { } } } + +// --------------------------------------------------------------------------- +// DataScienceCluster — minimal representation for DSC state inspection. +// Only spec.components.mlflowoperator.managementState is needed. +// --------------------------------------------------------------------------- + +type DataScienceCluster struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec DSCSpec `json:"spec,omitempty"` +} + +type DSCSpec struct { + Components DSCComponents `json:"components,omitempty"` +} + +type DSCComponents struct { + MLflowOperator DSCComponentState `json:"mlflowoperator,omitempty"` +} + +type DSCComponentState struct { + ManagementState string `json:"managementState,omitempty"` +} + +type DataScienceClusterList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []DataScienceCluster `json:"items"` +} + +func (in *DataScienceCluster) DeepCopyObject() runtime.Object { + return in.DeepCopy() +} + +func (in *DataScienceCluster) DeepCopy() *DataScienceCluster { + if in == nil { + return nil + } + out := new(DataScienceCluster) + in.DeepCopyInto(out) + return out +} + +func (in *DataScienceCluster) DeepCopyInto(out *DataScienceCluster) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + out.Spec = in.Spec +} + +func (in *DataScienceClusterList) DeepCopyObject() runtime.Object { + return in.DeepCopy() +} + +func (in *DataScienceClusterList) DeepCopy() *DataScienceClusterList { + if in == nil { + return nil + } + out := new(DataScienceClusterList) + in.DeepCopyInto(out) + return out +} + +func (in *DataScienceClusterList) DeepCopyInto(out *DataScienceClusterList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + out.Items = make([]DataScienceCluster, len(in.Items)) + for i := range in.Items { + in.Items[i].DeepCopyInto(&out.Items[i]) + } + } +} From f1156e06248b7d8134ccc8c4462830a41c3ef3f5 Mon Sep 17 00:00:00 2001 From: Bobbins228 Date: Wed, 27 May 2026 14:49:13 +0100 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20resolve=20OTel=E2=86=92MLflow=20trac?= =?UTF-8?q?e=20export=20failures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Set service-ca.crt as ca_file for MLflow TLS verification - Add mlflow.workspace and mlflow.experimentName Helm values for RHOAI workspace/experiment header configuration - Create MLflow experiment via API at bootstrap using configured name - Wire --mlflow-workspace and --mlflow-experiment-name CLI flags - Prefer mlflow-operator-mlflow-integration ClusterRole for gatewayendpoints/use permission Signed-off-by: Bobbins228 --- .../templates/manager/manager.yaml | 6 + charts/kagenti-operator/values.yaml | 5 + kagenti-operator/cmd/main.go | 18 ++- kagenti-operator/internal/bootstrap/otel.go | 150 +++++++++++++++--- .../internal/bootstrap/otel_test.go | 93 ++++++++++- .../internal/bootstrap/presets.go | 4 - .../controller/mlflow_operand_controller.go | 10 +- 7 files changed, 240 insertions(+), 46 deletions(-) diff --git a/charts/kagenti-operator/templates/manager/manager.yaml b/charts/kagenti-operator/templates/manager/manager.yaml index a0c2992d..03d4ec68 100644 --- a/charts/kagenti-operator/templates/manager/manager.yaml +++ b/charts/kagenti-operator/templates/manager/manager.yaml @@ -37,6 +37,12 @@ spec: {{- if .Values.otelBootstrap.enable }} - "--enable-otel-bootstrap=true" {{- end }} + {{- if .Values.mlflow.workspace }} + - "--mlflow-workspace={{ .Values.mlflow.workspace }}" + {{- end }} + {{- if .Values.mlflow.experimentName }} + - "--mlflow-experiment-name={{ .Values.mlflow.experimentName }}" + {{- end }} {{- if .Values.verifiedFetch.enabled }} - "--enable-verified-fetch=true" - "--verified-fetch-spiffe-socket={{ .Values.verifiedFetch.spiffeEndpointSocket }}" diff --git a/charts/kagenti-operator/values.yaml b/charts/kagenti-operator/values.yaml index 6968753e..352d1368 100644 --- a/charts/kagenti-operator/values.yaml +++ b/charts/kagenti-operator/values.yaml @@ -113,6 +113,11 @@ authbridgeConfig: # [MLFLOW]: MLflow experiment tracking integration mlflow: enable: false + # Kubernetes namespace used as the x-mlflow-workspace header value (RHOAI only). + # Required for RHOAI MLflow; leave empty for vanilla MLflow deployments. + workspace: "" + # MLflow experiment name. Created automatically if it doesn't exist. + experimentName: "kagenti-traces" # [OTEL BOOTSTRAP]: OTel collector bootstrap at operator startup. # When enabled, the operator assembles the OTel collector ConfigMap from diff --git a/kagenti-operator/cmd/main.go b/kagenti-operator/cmd/main.go index 46cb8e8a..9ceeb56e 100644 --- a/kagenti-operator/cmd/main.go +++ b/kagenti-operator/cmd/main.go @@ -105,6 +105,8 @@ func main() { var enforceNetworkPolicies bool var enableMLflow bool var enableOtelBootstrap bool + var mlflowWorkspace string + var mlflowExperimentName string var enableCardDiscovery bool @@ -158,6 +160,10 @@ func main() { "Enable MLflow experiment tracking integration") flag.BoolVar(&enableOtelBootstrap, "enable-otel-bootstrap", false, "Enable OTel collector bootstrap (ingress CA trust and ConfigMap assembly) at startup") + flag.StringVar(&mlflowWorkspace, "mlflow-workspace", "", + "Kubernetes namespace used as the x-mlflow-workspace header value (RHOAI only)") + flag.StringVar(&mlflowExperimentName, "mlflow-experiment-name", "kagenti-traces", + "MLflow experiment name; created automatically if it doesn't exist") flag.BoolVar(&enableCardDiscovery, "enable-card-discovery", false, "Enable automatic agent card discovery from AgentRuntime workloads into status.card") @@ -615,11 +621,13 @@ func main() { if enableOtelBootstrap { otelBootstrap := &bootstrap.OtelBootstrapRunnable{ - Client: mgr.GetClient(), - APIReader: mgr.GetAPIReader(), - Config: mgr.GetConfig(), - Namespace: getOperatorNamespace(), - Log: ctrl.Log.WithName("bootstrap"), + Client: mgr.GetClient(), + APIReader: mgr.GetAPIReader(), + Config: mgr.GetConfig(), + Namespace: getOperatorNamespace(), + Log: ctrl.Log.WithName("bootstrap"), + MLflowWorkspace: mlflowWorkspace, + MLflowExperimentName: mlflowExperimentName, } if err := mgr.Add(otelBootstrap); err != nil { setupLog.Error(err, "unable to add OTel bootstrap runnable") diff --git a/kagenti-operator/internal/bootstrap/otel.go b/kagenti-operator/internal/bootstrap/otel.go index 8f3017f0..5b4dd796 100644 --- a/kagenti-operator/internal/bootstrap/otel.go +++ b/kagenti-operator/internal/bootstrap/otel.go @@ -19,8 +19,12 @@ package bootstrap import ( "context" "crypto/sha256" + "crypto/tls" + "crypto/x509" "fmt" + "net/http" "net/url" + "os" "strings" "time" @@ -75,10 +79,18 @@ type OtelBootstrapRunnable struct { Namespace string Log logr.Logger + // MLflowWorkspace is the Kubernetes namespace used as the x-mlflow-workspace + // header value (RHOAI only). When empty, workspace/experiment headers are skipped. + MLflowWorkspace string + // MLflowExperimentName is the MLflow experiment name to create/lookup. + MLflowExperimentName string + // IsOpenShift overrides OCP detection when non-nil (for testing). IsOpenShift func(ctx context.Context) (bool, error) // MLflowCRDExists overrides CRD discovery when non-nil (for testing). MLflowCRDExists func(ctx context.Context) (bool, error) + // EnsureExperiment overrides MLflow experiment creation when non-nil (for testing). + EnsureExperiment func(ctx context.Context, baseURL, workspace string) (string, error) } // Start runs the bootstrap sequence. Called by the manager after leader election @@ -225,6 +237,25 @@ func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, lo return err } + if mf.available && isOCP && r.MLflowWorkspace != "" { + mf.workspaceNS = r.MLflowWorkspace + baseURL := strings.TrimSuffix(mf.tracesURL, "/v1/traces") + if baseURL != "" && baseURL != mf.tracesURL { + expID, expErr := r.ensureMLflowExperiment(ctx, log, baseURL, mf.workspaceNS) + if expErr != nil { + log.Error(expErr, "Could not create MLflow experiment, traces may not be routed correctly", + "workspace", mf.workspaceNS) + } else { + mf.experimentID = expID + log.Info("MLflow workspace and experiment ready", + "workspace", mf.workspaceNS, "experimentID", expID) + } + } + } else if mf.available && isOCP && r.MLflowWorkspace == "" { + log.Info("mlflow.workspace not configured, skipping workspace and experiment headers. " + + "Set mlflow.workspace in Helm values for RHOAI deployments.") + } + phoenixAvailable := r.discoverPhoenix(ctx, log) config, err := assembleCollectorConfig(isOCP, mf, phoenixAvailable) @@ -300,11 +331,12 @@ func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, lo return r.rolloutRestartCollector(ctx, log, configHash) } -// mlflowInfo holds the discovered MLflow endpoint and workspace namespace. +// mlflowInfo holds the discovered MLflow endpoint and workspace config. type mlflowInfo struct { - available bool - tracesURL string // in-cluster traces endpoint (e.g. https://mlflow.ns.svc:8443/v1/traces) - workspaceNS string // namespace for x-mlflow-workspace header + available bool + tracesURL string // in-cluster traces endpoint (e.g. https://mlflow.ns.svc:8443/v1/traces) + workspaceNS string // agent namespace for x-mlflow-workspace header (RHOAI only) + experimentID string // MLflow experiment ID for x-mlflow-experiment-id header } // discoverMLflow checks for the MLflow CRD and, if present, discovers the @@ -347,25 +379,17 @@ func (r *OtelBootstrapRunnable) discoverMLflow(ctx context.Context, log logr.Log return info, nil } -// mlflowInfoFromCR extracts the in-cluster endpoint and workspace namespace -// from an MLflow CR. The MLflow CRD is cluster-scoped, so cr.Namespace is -// always empty; we derive the namespace from status.address.url instead -// (e.g. "https://mlflow.redhat-ods-applications.svc:8443"). +// mlflowInfoFromCR extracts the in-cluster traces endpoint from an MLflow CR. func mlflowInfoFromCR(cr *mlflow.MLflow, log logr.Logger) *mlflowInfo { info := &mlflowInfo{available: true} if cr.Status.Address != nil && cr.Status.Address.URL != "" { parsed, err := url.Parse(cr.Status.Address.URL) if err == nil && parsed.Scheme != "" && parsed.Host != "" { - hostname := parsed.Hostname() - parts := strings.SplitN(hostname, ".", 3) - if len(parts) >= 2 { - info.workspaceNS = parts[1] - } info.tracesURL = fmt.Sprintf("%s://%s/v1/traces", parsed.Scheme, parsed.Host) log.Info("Found available MLflow CR", "name", cr.Name, "addressURL", cr.Status.Address.URL, - "tracesURL", info.tracesURL, "workspaceNS", info.workspaceNS) + "tracesURL", info.tracesURL) return info } log.Info("MLflow address URL missing scheme or host, falling back", @@ -456,6 +480,56 @@ func (r *OtelBootstrapRunnable) discoverPhoenix(ctx context.Context, log logr.Lo return true } +// ensureMLflowExperiment creates or retrieves the configured experiment in the +// given workspace and returns the experiment ID. +func (r *OtelBootstrapRunnable) ensureMLflowExperiment(ctx context.Context, log logr.Logger, baseURL, workspace string) (string, error) { + if r.EnsureExperiment != nil { + return r.EnsureExperiment(ctx, baseURL, workspace) + } + + httpClient, err := newServiceCAHTTPClient() + if err != nil { + return "", fmt.Errorf("creating TLS HTTP client: %w", err) + } + + c := &mlflow.Client{ + BaseURL: baseURL, + HTTPClient: httpClient, + } + + expName := r.MLflowExperimentName + if expName == "" { + expName = "kagenti-traces" + } + + expID, err := c.CreateExperiment(ctx, expName, workspace) + if err != nil { + return "", fmt.Errorf("creating MLflow experiment %q: %w", expName, err) + } + + log.Info("Created/found MLflow experiment", "name", expName, "workspace", workspace, "experimentID", expID) + return expID, nil +} + +// newServiceCAHTTPClient returns an HTTP client configured to trust the +// OpenShift service-serving CA certificate. +func newServiceCAHTTPClient() (*http.Client, error) { + caCert, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt") + if err != nil { + return nil, fmt.Errorf("reading service-ca.crt: %w", err) + } + pool := x509.NewCertPool() + pool.AppendCertsFromPEM(caCert) + return &http.Client{ + Timeout: 30 * time.Second, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{ + RootCAs: pool, + }, + }, + }, nil +} + // rolloutRestartCollector patches the OTel collector Deployment's pod template // annotation to trigger a rollout restart. func (r *OtelBootstrapRunnable) rolloutRestartCollector(ctx context.Context, log logr.Logger, configHash string) error { @@ -524,8 +598,12 @@ func assembleCollectorConfig(isOCP bool, mf *mlflowInfo, phoenixAvailable bool) } mergeDeep(config, rhoaiAuth) - clearMLflowExporterTLS(config) - setMLflowBearerTokenAuth(config, mf.workspaceNS) + setMLflowExporterServiceCATLS(config) + setMLflowBearerTokenAuth(config) + + if mf.workspaceNS != "" { + setMLflowHeaders(config, mf.workspaceNS, mf.experimentID) + } } else { mlflowAuth, err := parsePreset(mlflowAuthPreset) if err != nil { @@ -595,9 +673,11 @@ func mergeDeep(dst, src map[string]any) { } } -// clearMLflowExporterTLS clears the TLS config on the MLflow exporter for RHOAI -// (RHOAI uses service-ca.crt from the SA token projection). -func clearMLflowExporterTLS(config map[string]any) { +// setMLflowExporterServiceCATLS replaces the mlflow preset's tls.insecure config +// with the OpenShift service-serving CA cert used to verify the MLflow TLS +// certificate. Without this, the collector rejects the MLflow cert as +// "x509: certificate signed by unknown authority." +func setMLflowExporterServiceCATLS(config map[string]any) { exporters, ok := config["exporters"].(map[string]any) if !ok { return @@ -606,12 +686,15 @@ func clearMLflowExporterTLS(config map[string]any) { if !ok { return } - mlflowExp["tls"] = map[string]any{} + mlflowExp["tls"] = map[string]any{ + "ca_file": "/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt", + } } -// setMLflowBearerTokenAuth sets bearer token auth and workspace headers on the -// MLflow exporter for RHOAI deployments. -func setMLflowBearerTokenAuth(config map[string]any, mlflowNamespace string) { +// setMLflowHeaders sets the workspace and experiment-id headers on the MLflow +// exporter. RHOAI requires both: x-mlflow-workspace scopes to a namespace, and +// x-mlflow-experiment-id routes traces to a specific experiment. +func setMLflowHeaders(config map[string]any, workspace, experimentID string) { exporters, ok := config["exporters"].(map[string]any) if !ok { return @@ -620,14 +703,29 @@ func setMLflowBearerTokenAuth(config map[string]any, mlflowNamespace string) { if !ok { return } - mlflowExp["auth"] = map[string]any{"authenticator": "bearertokenauth/mlflow"} - headers, ok := mlflowExp["headers"].(map[string]any) if !ok { headers = map[string]any{} mlflowExp["headers"] = headers } - headers["x-mlflow-workspace"] = mlflowNamespace + headers["x-mlflow-workspace"] = workspace + if experimentID != "" { + headers["x-mlflow-experiment-id"] = experimentID + } +} + +// setMLflowBearerTokenAuth sets bearer token auth on the MLflow exporter for +// RHOAI deployments. +func setMLflowBearerTokenAuth(config map[string]any) { + exporters, ok := config["exporters"].(map[string]any) + if !ok { + return + } + mlflowExp, ok := exporters["otlphttp/mlflow"].(map[string]any) + if !ok { + return + } + mlflowExp["auth"] = map[string]any{"authenticator": "bearertokenauth/mlflow"} } // setMLflowOAuthAuth sets OAuth2 client authentication on the MLflow exporter diff --git a/kagenti-operator/internal/bootstrap/otel_test.go b/kagenti-operator/internal/bootstrap/otel_test.go index 6d5a6556..7f1c0e3b 100644 --- a/kagenti-operator/internal/bootstrap/otel_test.go +++ b/kagenti-operator/internal/bootstrap/otel_test.go @@ -302,7 +302,50 @@ func TestMLflowCRDPresent_OCP_UsesRHOAIAuth(t *testing.T) { config := cm.Data[configMapDataKey] assertContains(t, config, "bearertokenauth/mlflow", "Expected RHOAI bearer token auth on OCP") - assertContains(t, config, "x-mlflow-workspace", "Expected RHOAI workspace header on OCP") + assertNotContains(t, config, "x-mlflow-workspace", + "Workspace header should not be set when mlflow.workspace is not configured") + assertContains(t, config, "ca_file", "Expected service-ca.crt TLS config on OCP") +} + +func TestMLflowCRDPresent_OCP_WithWorkspace_SetsHeaders(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + cr := mlflowCR("mlflow", "redhat-ods-applications", true, "") + cr.Status.Address = &mlflow.MLflowAddress{URL: "https://mlflow.redhat-ods-applications.svc:8443"} + + ingressCertCM := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: ingressCertConfigMap, + Namespace: ingressCertNamespace, + }, + Data: map[string]string{caBundleKey: "-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----"}, + } + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(dep, cr, ingressCertCM).Build() + r := newRunnable(cl, isOpenShift, mlflowCRDPresent) + r.MLflowWorkspace = "team1" + r.MLflowExperimentName = "kagenti-traces" + r.EnsureExperiment = func(_ context.Context, _, _ string) (string, error) { + return "42", nil + } + + if err := r.Start(context.Background()); err != nil { + t.Fatalf("Start() failed: %v", err) + } + + cm := &corev1.ConfigMap{} + if err := cl.Get(context.Background(), types.NamespacedName{ + Name: collectorConfigMapName, Namespace: testNamespace, + }, cm); err != nil { + t.Fatalf("Expected collector ConfigMap to exist: %v", err) + } + + config := cm.Data[configMapDataKey] + assertContains(t, config, "x-mlflow-workspace", "Expected workspace header when workspace configured") + assertContains(t, config, "team1", "Expected workspace to match configured value") + assertContains(t, config, "x-mlflow-experiment-id", "Expected experiment-id header") + assertContains(t, config, "42", "Expected experiment ID from MLflow API") } func TestMLflowCRDPresent_NonOCP_UsesOAuthAuth(t *testing.T) { @@ -555,7 +598,7 @@ func TestAssembleConfig_DefaultOnly(t *testing.T) { } func TestAssembleConfig_PhoenixAndMLflow(t *testing.T) { - cfg, err := assembleCollectorConfig(false, &mlflowInfo{available: true, tracesURL: "http://mlflow.mlflow-ns.svc:5000/v1/traces", workspaceNS: "mlflow-ns"}, true) + cfg, err := assembleCollectorConfig(false, &mlflowInfo{available: true, tracesURL: "http://mlflow.mlflow-ns.svc:5000/v1/traces"}, true) if err != nil { t.Fatalf("assembleCollectorConfig failed: %v", err) } @@ -574,8 +617,8 @@ func TestAssembleConfig_PhoenixAndMLflow(t *testing.T) { } } -func TestAssembleConfig_OCP_MLflow_IngressCATLS(t *testing.T) { - cfg, err := assembleCollectorConfig(true, &mlflowInfo{available: true, tracesURL: "https://mlflow.rhoai-ns.svc:8443/v1/traces", workspaceNS: "rhoai-ns"}, false) +func TestAssembleConfig_OCP_MLflow_ServiceCATLS(t *testing.T) { + cfg, err := assembleCollectorConfig(true, &mlflowInfo{available: true, tracesURL: "https://mlflow.rhoai-ns.svc:8443/v1/traces"}, false) if err != nil { t.Fatalf("assembleCollectorConfig failed: %v", err) } @@ -599,12 +642,48 @@ func TestAssembleConfig_OCP_MLflow_IngressCATLS(t *testing.T) { t.Error("Expected bearertokenauth/mlflow authenticator on OCP") } + tlsCfg, ok := mlflowExp["tls"].(map[string]any) + if !ok { + t.Fatal("Expected tls config on MLflow exporter") + } + if tlsCfg["ca_file"] != "/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt" { + t.Errorf("Expected service-ca.crt ca_file, got %v", tlsCfg["ca_file"]) + } + if _, hasInsecure := tlsCfg["insecure"]; hasInsecure { + t.Error("Expected insecure flag to be cleared on OCP") + } + + // No workspace/experiment headers when mlflowInfo has no workspaceNS + if headers, ok := mlflowExp["headers"].(map[string]any); ok { + if _, hasWorkspace := headers["x-mlflow-workspace"]; hasWorkspace { + t.Error("x-mlflow-workspace should not be set when no workspace is configured") + } + } +} + +func TestAssembleConfig_OCP_MLflow_WithWorkspace(t *testing.T) { + cfg, err := assembleCollectorConfig(true, &mlflowInfo{ + available: true, + tracesURL: "https://mlflow.rhoai-ns.svc:8443/v1/traces", + workspaceNS: "team1", + experimentID: "5", + }, false) + if err != nil { + t.Fatalf("assembleCollectorConfig failed: %v", err) + } + + exporters := cfg["exporters"].(map[string]any) + mlflowExp := exporters["otlphttp/mlflow"].(map[string]any) + headers, ok := mlflowExp["headers"].(map[string]any) if !ok { - t.Fatal("Expected headers on MLflow exporter") + t.Fatal("Expected headers on MLflow exporter when workspace is set") + } + if headers["x-mlflow-workspace"] != "team1" { + t.Errorf("Expected x-mlflow-workspace=team1, got %v", headers["x-mlflow-workspace"]) } - if headers["x-mlflow-workspace"] != "rhoai-ns" { - t.Errorf("Expected workspace header 'rhoai-ns', got %v", headers["x-mlflow-workspace"]) + if headers["x-mlflow-experiment-id"] != "5" { + t.Errorf("Expected x-mlflow-experiment-id=5, got %v", headers["x-mlflow-experiment-id"]) } } diff --git a/kagenti-operator/internal/bootstrap/presets.go b/kagenti-operator/internal/bootstrap/presets.go index d2ffe252..86811ae8 100644 --- a/kagenti-operator/internal/bootstrap/presets.go +++ b/kagenti-operator/internal/bootstrap/presets.go @@ -113,8 +113,6 @@ exporters: traces_endpoint: http://mlflow:5000/v1/traces tls: insecure: true - headers: - x-mlflow-experiment-id: "0" retry_on_failure: enabled: true initial_interval: 5s @@ -161,8 +159,6 @@ exporters: compression: none retry_on_failure: enabled: false - tls: - ca_file: /var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt extensions: bearertokenauth/mlflow: filename: /var/run/secrets/kubernetes.io/serviceaccount/token diff --git a/kagenti-operator/internal/controller/mlflow_operand_controller.go b/kagenti-operator/internal/controller/mlflow_operand_controller.go index 38bf4f7a..2a08ff1a 100644 --- a/kagenti-operator/internal/controller/mlflow_operand_controller.go +++ b/kagenti-operator/internal/controller/mlflow_operand_controller.go @@ -65,7 +65,8 @@ const ( // MLflowClusterRoleIntegration is the ClusterRole created by the MLflow operator. MLflowClusterRoleIntegration = "mlflow-operator-mlflow-integration" - // MLflowClusterRoleEdit is the newer ClusterRole name (preferred). + // MLflowClusterRoleEdit is the newer ClusterRole name; lacks + // gatewayendpoints/use needed for OTLP trace ingestion. MLflowClusterRoleEdit = "mlflow-operator-mlflow-edit" // requeueMLflowNotReady is the requeue delay when MLflow is not yet available. @@ -328,14 +329,15 @@ func (r *MLflowOperandReconciler) ensureOTELRoleBinding(ctx context.Context, nam return nil } -// resolveMLflowClusterRole finds the MLflow ClusterRole, preferring the newer -// "edit" name, falling back to "integration". +// resolveMLflowClusterRole finds the MLflow ClusterRole, preferring +// "integration" which includes the gatewayendpoints/use verb required for +// OTLP trace ingestion. func (r *MLflowOperandReconciler) resolveMLflowClusterRole(ctx context.Context) string { if r.ResolveMLflowClusterRole != nil { return r.ResolveMLflowClusterRole(ctx) } - for _, name := range []string{MLflowClusterRoleEdit, MLflowClusterRoleIntegration} { + for _, name := range []string{MLflowClusterRoleIntegration, MLflowClusterRoleEdit} { cr := &rbacv1.ClusterRole{} if err := r.Get(ctx, types.NamespacedName{Name: name}, cr); err == nil { return name