From 4ff292e3165cf0dbf4de8c81f610d596261e8ec4 Mon Sep 17 00:00:00 2001 From: Bobbins228 Date: Mon, 25 May 2026 10:56:16 +0100 Subject: [PATCH 1/3] feat: add OTel collector bootstrap as startup Runnable Implement OtelBootstrapRunnable that replaces the Helm-based otel-ingress-ca-job and collectorConfig helper with operator-native startup logic: - Step 1: Project OpenShift ingress CA into kagenti-system (OCP only) - Step 2: Assemble OTel collector ConfigMap from detected components (Phoenix, MLflow) with dynamic CRD/service discovery Key behaviours: - OCP detection via config.openshift.io API group discovery - MLflow CRD-missing graceful degradation with restart guidance - MLflow service backoff retry when CRD present but CR not ready - Idempotent: no Deployment restart when ConfigMap unchanged - RHOAI bearer token auth on OCP, OAuth2 client auth elsewhere New files: internal/bootstrap/otel.go - Runnable implementation internal/bootstrap/presets.go - Ported Helm preset configs internal/bootstrap/otel_test.go - 16 unit tests Assisted-by: Cursor Signed-off-by: Bobbins228 --- .../templates/manager/manager.yaml | 3 + .../kagenti-operator/templates/rbac/role.yaml | 1 + charts/kagenti-operator/values.yaml | 8 + kagenti-operator/cmd/main.go | 19 + kagenti-operator/internal/bootstrap/otel.go | 599 ++++++++++++++++ .../internal/bootstrap/otel_test.go | 646 ++++++++++++++++++ .../internal/bootstrap/presets.go | 171 +++++ 7 files changed, 1447 insertions(+) create mode 100644 kagenti-operator/internal/bootstrap/otel.go create mode 100644 kagenti-operator/internal/bootstrap/otel_test.go create mode 100644 kagenti-operator/internal/bootstrap/presets.go diff --git a/charts/kagenti-operator/templates/manager/manager.yaml b/charts/kagenti-operator/templates/manager/manager.yaml index 6a208550..c80a1077 100644 --- a/charts/kagenti-operator/templates/manager/manager.yaml +++ b/charts/kagenti-operator/templates/manager/manager.yaml @@ -34,6 +34,9 @@ spec: {{- if .Values.mlflow.enable }} - "--enable-mlflow=true" {{- end }} + {{- if .Values.otelBootstrap.enable }} + - "--enable-otel-bootstrap=true" + {{- end }} {{- if .Values.verifiedFetch.enabled }} - "--enable-verified-fetch=true" - "--verified-fetch-spiffe-socket={{ .Values.verifiedFetch.spiffeEndpointSocket }}" diff --git a/charts/kagenti-operator/templates/rbac/role.yaml b/charts/kagenti-operator/templates/rbac/role.yaml index 16f4c4a5..d3c715e4 100755 --- a/charts/kagenti-operator/templates/rbac/role.yaml +++ b/charts/kagenti-operator/templates/rbac/role.yaml @@ -19,6 +19,7 @@ rules: - apiGroups: - "" resources: + - endpoints - namespaces - services verbs: diff --git a/charts/kagenti-operator/values.yaml b/charts/kagenti-operator/values.yaml index 2f66a0b2..57696a2c 100644 --- a/charts/kagenti-operator/values.yaml +++ b/charts/kagenti-operator/values.yaml @@ -96,6 +96,14 @@ enforceNetworkPolicies: false mlflow: enable: false +# [OTEL BOOTSTRAP]: OTel collector bootstrap at operator startup. +# When enabled, the operator assembles the OTel collector ConfigMap from +# detected components (Phoenix, MLflow) and projects the OpenShift ingress +# CA into the operator namespace (OCP only). Replaces the Helm-based +# otel-ingress-ca-job and kagenti.otel.collectorConfig helper. +otelBootstrap: + enable: false + # [VERIFIED FETCH]: mTLS-authenticated fetch of agent cards via SPIFFE identity (Phase 1) # When enabled, the operator uses go-spiffe mTLS to fetch agent cards and records # the agent's attested SPIFFE ID in CRD status. diff --git a/kagenti-operator/cmd/main.go b/kagenti-operator/cmd/main.go index 6ee3c106..3f25f6de 100644 --- a/kagenti-operator/cmd/main.go +++ b/kagenti-operator/cmd/main.go @@ -46,6 +46,7 @@ import ( agentv1alpha1 "github.com/kagenti/operator/api/v1alpha1" "github.com/kagenti/operator/internal/agentcard" + "github.com/kagenti/operator/internal/bootstrap" "github.com/kagenti/operator/internal/controller" "github.com/kagenti/operator/internal/keycloak" "github.com/kagenti/operator/internal/mlflow" @@ -100,6 +101,7 @@ func main() { var signatureAuditMode bool var enforceNetworkPolicies bool var enableMLflow bool + var enableOtelBootstrap bool var enableVerifiedFetch bool var verifiedFetchSpiffeSocket string @@ -141,6 +143,8 @@ func main() { "Create NetworkPolicies to restrict traffic for agents with unverified signatures") flag.BoolVar(&enableMLflow, "enable-mlflow", false, "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.BoolVar(&enableVerifiedFetch, "enable-verified-fetch", false, "Enable mTLS-authenticated fetch of agent cards via SPIFFE identity") @@ -502,6 +506,21 @@ func main() { } // +kubebuilder:scaffold:builder + if enableOtelBootstrap { + otelBootstrap := &bootstrap.OtelBootstrapRunnable{ + Client: mgr.GetClient(), + APIReader: mgr.GetAPIReader(), + Config: mgr.GetConfig(), + Namespace: getOperatorNamespace(), + Log: ctrl.Log.WithName("bootstrap"), + } + if err := mgr.Add(otelBootstrap); err != nil { + setupLog.Error(err, "unable to add OTel bootstrap runnable") + os.Exit(1) + } + setupLog.Info("OTel collector bootstrap enabled") + } + if metricsCertWatcher != nil { setupLog.Info("Adding metrics certificate watcher to manager") if err := mgr.Add(metricsCertWatcher); err != nil { diff --git a/kagenti-operator/internal/bootstrap/otel.go b/kagenti-operator/internal/bootstrap/otel.go new file mode 100644 index 00000000..aae93888 --- /dev/null +++ b/kagenti-operator/internal/bootstrap/otel.go @@ -0,0 +1,599 @@ +/* +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 bootstrap + +import ( + "context" + "crypto/sha256" + "fmt" + "time" + + "github.com/go-logr/logr" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/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/types" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/discovery" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/yaml" + + "github.com/kagenti/operator/internal/mlflow" +) + +const ( + ingressCAConfigMap = "otel-ingress-ca" + ingressCertConfigMap = "default-ingress-cert" + ingressCertNamespace = "openshift-config-managed" + rootCAConfigMap = "kube-root-ca.crt" + rootCANamespace = "openshift-config" + collectorConfigMapName = "otel-collector-config" + collectorDeployment = "otel-collector" + phoenixServiceName = "phoenix" + + configMapDataKey = "base.yaml" + caBundleKey = "ca-bundle.crt" + rootCAKey = "ca.crt" + restartAnnotation = "kagenti.io/otel-config-hash" + ocpAPIGroup = "config.openshift.io" + mlflowCRDGroup = "mlflow.opendatahub.io" + mlflowCRDVersion = "v1" + mlflowCRDResource = "mlflows" + + defaultBackoffInitial = 5 * time.Second + defaultBackoffMax = 30 * time.Second + defaultBackoffTimeout = 5 * time.Minute +) + +// OtelBootstrapRunnable implements manager.Runnable. It runs once at operator +// startup to bootstrap the OTel collector infrastructure: +// - Step 1: Project the OpenShift ingress CA into the operator namespace (OCP only). +// - Step 2: Assemble the OTel collector ConfigMap from component-specific presets. +type OtelBootstrapRunnable struct { + Client client.Client + APIReader client.Reader + Config *rest.Config + Namespace string + Log logr.Logger + + // 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) +} + +// Start runs the bootstrap sequence. Called by the manager after leader election +// and cache sync, before controllers start processing events. +func (r *OtelBootstrapRunnable) Start(ctx context.Context) error { + log := r.Log.WithName("otel-bootstrap") + log.Info("Starting OTel collector bootstrap") + + isOCP, err := r.detectOpenShift(ctx) + if err != nil { + return fmt.Errorf("detecting OpenShift: %w", err) + } + + if isOCP { + log.Info("OpenShift detected, reconciling ingress CA trust") + if err := r.reconcileIngressCA(ctx, log); err != nil { + return fmt.Errorf("ingress CA bootstrap: %w", err) + } + } else { + log.Info("Not running on OpenShift, skipping ingress CA trust") + } + + if err := r.reconcileCollectorConfig(ctx, log, isOCP); err != nil { + return fmt.Errorf("collector config bootstrap: %w", err) + } + + log.Info("OTel collector bootstrap complete") + return nil +} + +// NeedLeaderElection returns true so the bootstrap only runs on the leader. +func (r *OtelBootstrapRunnable) NeedLeaderElection() bool { + return true +} + +// detectOpenShift checks for the config.openshift.io API group. +func (r *OtelBootstrapRunnable) detectOpenShift(ctx context.Context) (bool, error) { + if r.IsOpenShift != nil { + return r.IsOpenShift(ctx) + } + + dc, err := discovery.NewDiscoveryClientForConfig(r.Config) + if err != nil { + return false, fmt.Errorf("creating discovery client: %w", err) + } + + _, apiLists, err := dc.ServerGroupsAndResources() + if err != nil && !discovery.IsGroupDiscoveryFailedError(err) { + return false, fmt.Errorf("discovering API groups: %w", err) + } + + for _, list := range apiLists { + gv, parseErr := parseGroupVersion(list.GroupVersion) + if parseErr != nil { + continue + } + if gv == ocpAPIGroup { + return true, nil + } + } + return false, nil +} + +// reconcileIngressCA reads the OpenShift ingress CA and root CA, then creates +// or updates the otel-ingress-ca ConfigMap in the operator namespace. +func (r *OtelBootstrapRunnable) reconcileIngressCA(ctx context.Context, log logr.Logger) error { + ingressCert := &corev1.ConfigMap{} + key := types.NamespacedName{Name: ingressCertConfigMap, Namespace: ingressCertNamespace} + if err := r.APIReader.Get(ctx, key, ingressCert); err != nil { + return fmt.Errorf("reading %s/%s: %w", ingressCertNamespace, ingressCertConfigMap, err) + } + + caBundle, ok := ingressCert.Data[caBundleKey] + if !ok || caBundle == "" { + return fmt.Errorf("ConfigMap %s/%s has no %q key", ingressCertNamespace, ingressCertConfigMap, caBundleKey) + } + + rootCA := &corev1.ConfigMap{} + rootKey := types.NamespacedName{Name: rootCAConfigMap, Namespace: rootCANamespace} + if err := r.APIReader.Get(ctx, rootKey, rootCA); err != nil { + if !errors.IsNotFound(err) { + log.Info("Could not read root CA ConfigMap, continuing with ingress CA only", + "error", err, "configmap", rootCAConfigMap, "namespace", rootCANamespace) + } + } else if rootCert, ok := rootCA.Data[rootCAKey]; ok && rootCert != "" { + log.Info("Adding root CA to bundle for full certificate chain") + caBundle = caBundle + "\n" + rootCert + } + + existing := &corev1.ConfigMap{} + existingKey := types.NamespacedName{Name: ingressCAConfigMap, Namespace: r.Namespace} + if err := r.Client.Get(ctx, existingKey, existing); err != nil { + if !errors.IsNotFound(err) { + return fmt.Errorf("checking existing %s ConfigMap: %w", ingressCAConfigMap, err) + } + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: ingressCAConfigMap, + Namespace: r.Namespace, + Labels: map[string]string{ + "app.kubernetes.io/managed-by": "kagenti-operator", + "app.kubernetes.io/component": "otel-bootstrap", + }, + }, + Data: map[string]string{caBundleKey: caBundle}, + } + if err := r.Client.Create(ctx, cm); err != nil { + return fmt.Errorf("creating %s ConfigMap: %w", ingressCAConfigMap, err) + } + log.Info("Created ingress CA ConfigMap", "name", ingressCAConfigMap) + return nil + } + + if existing.Data[caBundleKey] == caBundle { + log.Info("Ingress CA ConfigMap already up-to-date, skipping") + return nil + } + + existing.Data = map[string]string{caBundleKey: caBundle} + if err := r.Client.Update(ctx, existing); err != nil { + return fmt.Errorf("updating %s ConfigMap: %w", ingressCAConfigMap, err) + } + log.Info("Updated ingress CA ConfigMap", "name", ingressCAConfigMap) + return nil +} + +// reconcileCollectorConfig discovers available components and assembles the +// OTel collector ConfigMap from preset configurations. +func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, log logr.Logger, isOCP bool) error { + mlflowAvailable, mlflowNamespace, err := r.discoverMLflow(ctx, log) + if err != nil { + return err + } + + phoenixAvailable := r.discoverPhoenix(ctx, log) + + config, err := assembleCollectorConfig(isOCP, mlflowAvailable, mlflowNamespace, phoenixAvailable) + if err != nil { + return fmt.Errorf("assembling collector config: %w", err) + } + + configYAML, err := yaml.Marshal(config) + if err != nil { + return fmt.Errorf("marshalling collector config: %w", err) + } + + configStr := string(configYAML) + configHash := fmt.Sprintf("%x", sha256.Sum256(configYAML)) + + existing := &corev1.ConfigMap{} + key := types.NamespacedName{Name: collectorConfigMapName, Namespace: r.Namespace} + if err := r.Client.Get(ctx, key, existing); err != nil { + if !errors.IsNotFound(err) { + return fmt.Errorf("checking existing collector ConfigMap: %w", err) + } + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: collectorConfigMapName, + Namespace: r.Namespace, + Labels: map[string]string{ + "app.kubernetes.io/managed-by": "kagenti-operator", + "app.kubernetes.io/component": "otel-bootstrap", + }, + Annotations: map[string]string{ + restartAnnotation: configHash, + }, + }, + Data: map[string]string{configMapDataKey: configStr}, + } + if err := r.Client.Create(ctx, cm); err != nil { + return fmt.Errorf("creating collector ConfigMap: %w", err) + } + log.Info("Created OTel collector ConfigMap", "components", + componentSummary(mlflowAvailable, phoenixAvailable)) + return r.rolloutRestartCollector(ctx, log, configHash) + } + + existingHash := existing.Annotations[restartAnnotation] + if existingHash == configHash { + log.Info("OTel collector ConfigMap unchanged, skipping restart") + return nil + } + + existing.Data = map[string]string{configMapDataKey: configStr} + if existing.Annotations == nil { + existing.Annotations = make(map[string]string) + } + existing.Annotations[restartAnnotation] = configHash + if err := r.Client.Update(ctx, existing); err != nil { + return fmt.Errorf("updating collector ConfigMap: %w", err) + } + log.Info("Updated OTel collector ConfigMap", "components", + componentSummary(mlflowAvailable, phoenixAvailable)) + + return r.rolloutRestartCollector(ctx, log, configHash) +} + +// discoverMLflow checks for the MLflow CRD and, if present, discovers the +// MLflow CR to derive the namespace where the service runs. +func (r *OtelBootstrapRunnable) discoverMLflow(ctx context.Context, log logr.Logger) (available bool, namespace string, err error) { + crdExists, err := r.mlflowCRDPresent(ctx) + if err != nil { + return false, "", fmt.Errorf("checking MLflow CRD: %w", err) + } + if !crdExists { + log.Info("MLflow CRD (mlflows.mlflow.opendatahub.io) not found. " + + "MLflow presets will be skipped. If the MLflow operator is installed later, " + + "restart the kagenti-operator pod to pick up MLflow configuration.") + return false, "", nil + } + + log.Info("MLflow CRD detected, discovering MLflow CR") + + list := &mlflow.MLflowList{} + if err := r.Client.List(ctx, list); err != nil { + log.Info("Could not list MLflow CRs, skipping MLflow presets", "error", err) + return false, "", nil + } + + for i := range list.Items { + cr := &list.Items[i] + if meta.IsStatusConditionTrue(cr.Status.Conditions, "Available") { + log.Info("Found available MLflow CR", + "name", cr.Name, "namespace", cr.Namespace, "url", cr.Status.URL) + return true, cr.Namespace, nil + } + } + + log.Info("MLflow CRD present but no Available MLflow CR found, waiting for service readiness") + available, namespace, err = r.waitForMLflowService(ctx, log) + if err != nil { + log.Info("MLflow service did not become ready within timeout, skipping MLflow presets", + "error", err) + return false, "", nil + } + return available, namespace, nil +} + +// waitForMLflowService retries with backoff until an MLflow CR becomes Available. +func (r *OtelBootstrapRunnable) waitForMLflowService(ctx context.Context, log logr.Logger) (bool, string, error) { + var resultAvailable bool + var resultNamespace string + + backoff := wait.Backoff{ + Duration: defaultBackoffInitial, + Factor: 2.0, + Cap: defaultBackoffMax, + Steps: 20, + } + + timeoutCtx, cancel := context.WithTimeout(ctx, defaultBackoffTimeout) + defer cancel() + + err := wait.ExponentialBackoffWithContext(timeoutCtx, backoff, func(ctx context.Context) (bool, error) { + list := &mlflow.MLflowList{} + if err := r.Client.List(ctx, list); err != nil { + log.V(1).Info("Retrying MLflow CR list", "error", err) + return false, nil + } + for i := range list.Items { + cr := &list.Items[i] + if meta.IsStatusConditionTrue(cr.Status.Conditions, "Available") { + log.Info("MLflow CR became Available", + "name", cr.Name, "namespace", cr.Namespace) + resultAvailable = true + resultNamespace = cr.Namespace + return true, nil + } + } + log.V(1).Info("MLflow CR not yet Available, retrying") + return false, nil + }) + + return resultAvailable, resultNamespace, err +} + +// mlflowCRDPresent checks if the mlflows.mlflow.opendatahub.io CRD is installed. +func (r *OtelBootstrapRunnable) mlflowCRDPresent(ctx context.Context) (bool, error) { + if r.MLflowCRDExists != nil { + return r.MLflowCRDExists(ctx) + } + + dc, err := discovery.NewDiscoveryClientForConfig(r.Config) + if err != nil { + return false, fmt.Errorf("creating discovery client: %w", err) + } + + resources, err := dc.ServerResourcesForGroupVersion(mlflowCRDGroup + "/" + mlflowCRDVersion) + if err != nil { + return false, nil //nolint:nilerr // CRD not installed is not an error + } + for _, res := range resources.APIResources { + if res.Name == mlflowCRDResource { + return true, nil + } + } + return false, nil +} + +// discoverPhoenix checks if the Phoenix service exists in the operator namespace. +func (r *OtelBootstrapRunnable) discoverPhoenix(ctx context.Context, log logr.Logger) bool { + svc := &corev1.Service{} + key := types.NamespacedName{Name: phoenixServiceName, Namespace: r.Namespace} + if err := r.Client.Get(ctx, key, svc); err != nil { + log.V(1).Info("Phoenix service not found, skipping Phoenix preset", + "namespace", r.Namespace) + return false + } + log.Info("Phoenix service detected", "namespace", r.Namespace) + return true +} + +// 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 { + dep := &appsv1.Deployment{} + key := types.NamespacedName{Name: collectorDeployment, Namespace: r.Namespace} + if err := r.Client.Get(ctx, key, dep); err != nil { + if errors.IsNotFound(err) { + log.Info("OTel collector Deployment not found, skipping rollout restart") + return nil + } + return fmt.Errorf("reading OTel collector Deployment: %w", err) + } + + if dep.Spec.Template.Annotations == nil { + dep.Spec.Template.Annotations = make(map[string]string) + } + + if dep.Spec.Template.Annotations[restartAnnotation] == configHash { + log.Info("OTel collector Deployment already has current config hash, skipping restart") + return nil + } + + dep.Spec.Template.Annotations[restartAnnotation] = configHash + if err := r.Client.Update(ctx, dep); err != nil { + return fmt.Errorf("rollout-restarting OTel collector: %w", err) + } + log.Info("Triggered OTel collector rollout restart", "configHash", configHash) + return nil +} + +// assembleCollectorConfig builds the complete OTel collector YAML config by +// merging the base config with component-specific presets. +func assembleCollectorConfig(isOCP, mlflowAvailable bool, mlflowNamespace string, phoenixAvailable bool) (map[string]any, error) { + config, err := parsePreset(baseConfig) + if err != nil { + return nil, fmt.Errorf("parsing base config: %w", err) + } + + hasComponentPipeline := false + + if phoenixAvailable { + phoenix, err := parsePreset(phoenixPreset) + if err != nil { + return nil, fmt.Errorf("parsing phoenix preset: %w", err) + } + mergeDeep(config, phoenix) + hasComponentPipeline = true + } + + if mlflowAvailable { + mlflowCfg, err := parsePreset(mlflowPreset) + if err != nil { + return nil, fmt.Errorf("parsing mlflow preset: %w", err) + } + mergeDeep(config, mlflowCfg) + hasComponentPipeline = true + + if isOCP { + rhoaiAuth, err := parsePreset(rhoaiMlflowAuthPreset) + if err != nil { + return nil, fmt.Errorf("parsing rhoai mlflow auth preset: %w", err) + } + mergeDeep(config, rhoaiAuth) + + clearMLflowExporterTLS(config) + setMLflowBearerTokenAuth(config, mlflowNamespace) + } else { + mlflowAuth, err := parsePreset(mlflowAuthPreset) + if err != nil { + return nil, fmt.Errorf("parsing mlflow auth preset: %w", err) + } + mergeDeep(config, mlflowAuth) + + setMLflowOAuthAuth(config) + } + } + + if !hasComponentPipeline { + defaultCfg, err := parsePreset(defaultPreset) + if err != nil { + return nil, fmt.Errorf("parsing default preset: %w", err) + } + mergeDeep(config, defaultCfg) + } + + if isOCP && mlflowAvailable { + setIngressCATLS(config) + } + + return config, nil +} + +// parsePreset unmarshals a YAML string into a map. +func parsePreset(yamlStr string) (map[string]any, error) { + var result map[string]any + if err := yaml.Unmarshal([]byte(yamlStr), &result); err != nil { + return nil, err + } + return result, nil +} + +// mergeDeep recursively merges src into dst. Maps are merged recursively; +// all other values (including slices) in src overwrite dst. +func mergeDeep(dst, src map[string]any) { + for key, srcVal := range src { + dstVal, exists := dst[key] + if !exists { + dst[key] = srcVal + continue + } + + dstMap, dstOk := dstVal.(map[string]any) + srcMap, srcOk := srcVal.(map[string]any) + if dstOk && srcOk { + mergeDeep(dstMap, srcMap) + } else { + dst[key] = srcVal + } + } +} + +// 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) { + exporters, ok := config["exporters"].(map[string]any) + if !ok { + return + } + mlflowExp, ok := exporters["otlphttp/mlflow"].(map[string]any) + if !ok { + return + } + mlflowExp["tls"] = map[string]any{} +} + +// setMLflowBearerTokenAuth sets bearer token auth and workspace headers on the +// MLflow exporter for RHOAI deployments. +func setMLflowBearerTokenAuth(config map[string]any, mlflowNamespace string) { + 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"} + + headers, ok := mlflowExp["headers"].(map[string]any) + if !ok { + headers = map[string]any{} + mlflowExp["headers"] = headers + } + headers["x-mlflow-workspace"] = mlflowNamespace +} + +// setMLflowOAuthAuth sets OAuth2 client authentication on the MLflow exporter +// for self-managed MLflow deployments. +func setMLflowOAuthAuth(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": "oauth2client/mlflow"} +} + +// setIngressCATLS sets the ingress CA file path on the OAuth2 client extension +// for OpenShift deployments. +func setIngressCATLS(config map[string]any) { + extensions, ok := config["extensions"].(map[string]any) + if !ok { + return + } + oauth, ok := extensions["oauth2client/mlflow"].(map[string]any) + if !ok { + return + } + oauth["tls"] = map[string]any{"ca_file": "/etc/pki/ingress-ca/ingress-ca.pem"} +} + +// componentSummary returns a human-readable summary of detected components. +func componentSummary(mlflow, phoenix bool) string { + s := "" + if mlflow { + s += "mlflow " + } + if phoenix { + s += "phoenix " + } + if s == "" { + s = "default (no component pipelines)" + } + return s +} + +// parseGroupVersion extracts the group from a "group/version" string. +func parseGroupVersion(gv string) (string, error) { + for i := len(gv) - 1; i >= 0; i-- { + if gv[i] == '/' { + return gv[:i], nil + } + } + return "", fmt.Errorf("no slash in group/version %q", gv) +} diff --git a/kagenti-operator/internal/bootstrap/otel_test.go b/kagenti-operator/internal/bootstrap/otel_test.go new file mode 100644 index 00000000..640c2c1e --- /dev/null +++ b/kagenti-operator/internal/bootstrap/otel_test.go @@ -0,0 +1,646 @@ +/* +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 bootstrap + +import ( + "context" + "testing" + + "github.com/go-logr/logr" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/log/zap" + + "github.com/kagenti/operator/internal/mlflow" +) + +const testNamespace = "kagenti-system" + +func testScheme() *runtime.Scheme { + s := runtime.NewScheme() + _ = corev1.AddToScheme(s) + _ = appsv1.AddToScheme(s) + _ = mlflow.AddToScheme(s) + return s +} + +func testLogger() logr.Logger { + return zap.New(zap.UseDevMode(true)) +} + +func newRunnable(cl client.Client, isOCP func(context.Context) (bool, error), mlflowCRD func(context.Context) (bool, error)) *OtelBootstrapRunnable { + return &OtelBootstrapRunnable{ + Client: cl, + APIReader: cl, + Namespace: testNamespace, + Log: testLogger(), + IsOpenShift: isOCP, + MLflowCRDExists: mlflowCRD, + } +} + +func otelCollectorDeployment() *appsv1.Deployment { + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: collectorDeployment, + Namespace: testNamespace, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app": "otel-collector"}, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"app": "otel-collector"}, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + {Name: "otel-collector", Image: "otel/opentelemetry-collector-contrib:0.122.1"}, + }, + }, + }, + }, + } +} + +func mlflowCR(name, namespace string, available bool, url string) *mlflow.MLflow { + conditions := []metav1.Condition{} + if available { + conditions = append(conditions, metav1.Condition{ + Type: "Available", + Status: metav1.ConditionTrue, + LastTransitionTime: metav1.Now(), + Reason: "Ready", + }) + } + return &mlflow.MLflow{ + TypeMeta: metav1.TypeMeta{ + APIVersion: "mlflow.opendatahub.io/v1", + Kind: "MLflow", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + }, + Status: mlflow.MLflowStatus{ + Conditions: conditions, + URL: url, + }, + } +} + +// --- OCP detection tests --- + +func TestStart_NonOCP_SkipsIngressCA(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(dep).Build() + r := newRunnable(cl, notOpenShift, noMLflowCRD) + + if err := r.Start(context.Background()); err != nil { + t.Fatalf("Start() failed: %v", err) + } + + cm := &corev1.ConfigMap{} + err := cl.Get(context.Background(), types.NamespacedName{ + Name: ingressCAConfigMap, Namespace: testNamespace, + }, cm) + if err == nil { + t.Error("Expected ingress CA ConfigMap to not exist on non-OCP, but it was created") + } +} + +func TestStart_OCP_CreatesIngressCA(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + + ingressCertCM := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: ingressCertConfigMap, + Namespace: ingressCertNamespace, + }, + Data: map[string]string{caBundleKey: "-----BEGIN CERTIFICATE-----\nINGRESS\n-----END CERTIFICATE-----"}, + } + rootCACM := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: rootCAConfigMap, + Namespace: rootCANamespace, + }, + Data: map[string]string{rootCAKey: "-----BEGIN CERTIFICATE-----\nROOT\n-----END CERTIFICATE-----"}, + } + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(dep, ingressCertCM, rootCACM).Build() + r := newRunnable(cl, isOpenShift, noMLflowCRD) + + 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: ingressCAConfigMap, Namespace: testNamespace, + }, cm); err != nil { + t.Fatalf("Expected ingress CA ConfigMap to exist: %v", err) + } + + bundle := cm.Data[caBundleKey] + if bundle == "" { + t.Fatal("Ingress CA ConfigMap has empty ca-bundle.crt") + } + if len(bundle) <= 50 { + t.Error("Expected concatenated ingress + root CA bundle") + } +} + +func TestIngressCA_Idempotent_NoUpdateWhenUnchanged(t *testing.T) { + scheme := testScheme() + + caBundle := "-----BEGIN CERTIFICATE-----\nINGRESS\n-----END CERTIFICATE-----" + ingressCertCM := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: ingressCertConfigMap, + Namespace: ingressCertNamespace, + }, + Data: map[string]string{caBundleKey: caBundle}, + } + existingCA := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: ingressCAConfigMap, + Namespace: testNamespace, + }, + Data: map[string]string{caBundleKey: caBundle}, + } + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(ingressCertCM, existingCA, otelCollectorDeployment()).Build() + r := newRunnable(cl, isOpenShift, noMLflowCRD) + + 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: ingressCAConfigMap, Namespace: testNamespace, + }, cm); err != nil { + t.Fatalf("Failed to get CA ConfigMap: %v", err) + } + if cm.Data[caBundleKey] != caBundle { + t.Error("CA bundle was modified when it should have been unchanged") + } +} + +// --- MLflow CRD missing tests --- + +func TestMLflowCRDMissing_SkipsMLflowPresets(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(dep).Build() + r := newRunnable(cl, notOpenShift, noMLflowCRD) + + 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] + if config == "" { + t.Fatal("Collector config is empty") + } + + assertContains(t, config, "traces/default", "Expected default pipeline when MLflow CRD missing") + assertNotContains(t, config, "otlphttp/mlflow", "Expected no MLflow exporter when CRD missing") +} + +// --- MLflow service discovery tests --- + +func TestMLflowCRDPresent_DiscoversCRNamespace(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + cr := mlflowCR("mlflow", "redhat-ods-applications", true, "https://mlflow.example.com") + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(dep, cr).Build() + r := newRunnable(cl, notOpenShift, mlflowCRDPresent) + + 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, "otlphttp/mlflow", "Expected MLflow exporter in config") + assertContains(t, config, "traces/mlflow", "Expected MLflow pipeline in config") +} + +func TestMLflowCRDPresent_OCP_UsesRHOAIAuth(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + cr := mlflowCR("mlflow", "redhat-ods-applications", true, "https://mlflow.example.com") + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(dep, cr).Build() + r := newRunnable(cl, isOpenShift, mlflowCRDPresent) + + ingressCertCM := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: ingressCertConfigMap, + Namespace: ingressCertNamespace, + }, + Data: map[string]string{caBundleKey: "-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----"}, + } + if err := cl.Create(context.Background(), ingressCertCM); err != nil { + t.Fatalf("Failed to create ingress cert: %v", err) + } + + 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, "bearertokenauth/mlflow", "Expected RHOAI bearer token auth on OCP") + assertContains(t, config, "x-mlflow-workspace", "Expected RHOAI workspace header on OCP") +} + +func TestMLflowCRDPresent_NonOCP_UsesOAuthAuth(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + cr := mlflowCR("mlflow", "default", true, "http://mlflow:5000") + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(dep, cr).Build() + r := newRunnable(cl, notOpenShift, mlflowCRDPresent) + + 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, "oauth2client/mlflow", "Expected OAuth2 client auth on non-OCP") +} + +// --- Phoenix tests --- + +func TestPhoenixPresent_MergesPhoenixPreset(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + phoenixSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: phoenixServiceName, + Namespace: testNamespace, + }, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{{Port: 4317}}, + }, + } + + cl := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(dep, phoenixSvc).Build() + r := newRunnable(cl, notOpenShift, noMLflowCRD) + + 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, "otlp/phoenix", "Expected Phoenix exporter in config") + assertContains(t, config, "traces/phoenix", "Expected Phoenix pipeline in config") + assertNotContains(t, config, "traces/default", "Expected no default pipeline when Phoenix is active") +} + +func TestPhoenixAbsent_NoPhoenixPreset(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(dep).Build() + r := newRunnable(cl, notOpenShift, noMLflowCRD) + + 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] + assertNotContains(t, config, "otlp/phoenix", "Expected no Phoenix exporter when service absent") +} + +// --- ConfigMap diff detection / idempotency tests --- + +func TestConfigMapUnchanged_NoDeploymentRestart(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(dep).Build() + r := newRunnable(cl, notOpenShift, noMLflowCRD) + + // First run: creates ConfigMap and restarts + if err := r.Start(context.Background()); err != nil { + t.Fatalf("First Start() failed: %v", err) + } + + // Capture the annotation after first run + updatedDep := &appsv1.Deployment{} + if err := cl.Get(context.Background(), types.NamespacedName{ + Name: collectorDeployment, Namespace: testNamespace, + }, updatedDep); err != nil { + t.Fatalf("Failed to get deployment: %v", err) + } + firstHash := updatedDep.Spec.Template.Annotations[restartAnnotation] + if firstHash == "" { + t.Fatal("Expected config hash annotation on Deployment after first run") + } + + // Second run: should detect no change + r2 := newRunnable(cl, notOpenShift, noMLflowCRD) + if err := r2.Start(context.Background()); err != nil { + t.Fatalf("Second Start() failed: %v", err) + } + + updatedDep2 := &appsv1.Deployment{} + if err := cl.Get(context.Background(), types.NamespacedName{ + Name: collectorDeployment, Namespace: testNamespace, + }, updatedDep2); err != nil { + t.Fatalf("Failed to get deployment: %v", err) + } + secondHash := updatedDep2.Spec.Template.Annotations[restartAnnotation] + if firstHash != secondHash { + t.Errorf("Config hash changed between idempotent runs: %s != %s", firstHash, secondHash) + } +} + +func TestConfigMapChanged_TriggersRestart(t *testing.T) { + scheme := testScheme() + dep := otelCollectorDeployment() + + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(dep).Build() + + // First run without Phoenix + r1 := newRunnable(cl, notOpenShift, noMLflowCRD) + if err := r1.Start(context.Background()); err != nil { + t.Fatalf("First Start() failed: %v", err) + } + + updatedDep := &appsv1.Deployment{} + _ = cl.Get(context.Background(), types.NamespacedName{ + Name: collectorDeployment, Namespace: testNamespace, + }, updatedDep) + firstHash := updatedDep.Spec.Template.Annotations[restartAnnotation] + + // Add Phoenix service + phoenixSvc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: phoenixServiceName, + Namespace: testNamespace, + }, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{{Port: 4317}}, + }, + } + if err := cl.Create(context.Background(), phoenixSvc); err != nil { + t.Fatalf("Failed to create Phoenix service: %v", err) + } + + // Second run with Phoenix: config changes, triggers restart + r2 := newRunnable(cl, notOpenShift, noMLflowCRD) + if err := r2.Start(context.Background()); err != nil { + t.Fatalf("Second Start() failed: %v", err) + } + + updatedDep2 := &appsv1.Deployment{} + _ = cl.Get(context.Background(), types.NamespacedName{ + Name: collectorDeployment, Namespace: testNamespace, + }, updatedDep2) + secondHash := updatedDep2.Spec.Template.Annotations[restartAnnotation] + + if firstHash == secondHash { + t.Error("Expected config hash to change when Phoenix becomes available") + } +} + +// --- assembleCollectorConfig unit tests --- + +func TestAssembleConfig_DefaultOnly(t *testing.T) { + cfg, err := assembleCollectorConfig(false, false, "", false) + if err != nil { + t.Fatalf("assembleCollectorConfig failed: %v", err) + } + + svc, ok := cfg["service"].(map[string]any) + if !ok { + t.Fatal("Expected 'service' key in config") + } + pipelines, ok := svc["pipelines"].(map[string]any) + if !ok { + t.Fatal("Expected 'pipelines' in service") + } + if _, ok := pipelines["traces/default"]; !ok { + t.Error("Expected traces/default pipeline") + } +} + +func TestAssembleConfig_PhoenixAndMLflow(t *testing.T) { + cfg, err := assembleCollectorConfig(false, true, "mlflow-ns", true) + if err != nil { + t.Fatalf("assembleCollectorConfig failed: %v", err) + } + + svc := cfg["service"].(map[string]any) + pipelines := svc["pipelines"].(map[string]any) + + if _, ok := pipelines["traces/phoenix"]; !ok { + t.Error("Expected traces/phoenix pipeline") + } + if _, ok := pipelines["traces/mlflow"]; !ok { + t.Error("Expected traces/mlflow pipeline") + } + if _, ok := pipelines["traces/default"]; ok { + t.Error("Expected no traces/default pipeline when components are active") + } +} + +func TestAssembleConfig_OCP_MLflow_IngressCATLS(t *testing.T) { + cfg, err := assembleCollectorConfig(true, true, "rhoai-ns", false) + if err != nil { + t.Fatalf("assembleCollectorConfig failed: %v", err) + } + + extensions := cfg["extensions"].(map[string]any) + bearer, ok := extensions["bearertokenauth/mlflow"].(map[string]any) + if !ok { + t.Fatal("Expected bearertokenauth/mlflow extension on OCP") + } + if bearer["filename"] != "/var/run/secrets/kubernetes.io/serviceaccount/token" { + t.Error("Expected SA token filename for bearer auth") + } + + exporters := cfg["exporters"].(map[string]any) + mlflowExp := exporters["otlphttp/mlflow"].(map[string]any) + auth, ok := mlflowExp["auth"].(map[string]any) + if !ok { + t.Fatal("Expected auth on MLflow exporter") + } + if auth["authenticator"] != "bearertokenauth/mlflow" { + t.Error("Expected bearertokenauth/mlflow authenticator on OCP") + } + + headers, ok := mlflowExp["headers"].(map[string]any) + if !ok { + t.Fatal("Expected headers on MLflow exporter") + } + if headers["x-mlflow-workspace"] != "rhoai-ns" { + t.Errorf("Expected workspace header 'rhoai-ns', got %v", headers["x-mlflow-workspace"]) + } +} + +// --- mergeDeep tests --- + +func TestMergeDeep_Basic(t *testing.T) { + dst := map[string]any{ + "a": "1", + "b": map[string]any{"c": "2"}, + } + src := map[string]any{ + "a": "overwritten", + "b": map[string]any{"d": "3"}, + "e": "new", + } + mergeDeep(dst, src) + + if dst["a"] != "overwritten" { + t.Error("Expected 'a' to be overwritten") + } + b := dst["b"].(map[string]any) + if b["c"] != "2" { + t.Error("Expected 'b.c' to be preserved") + } + if b["d"] != "3" { + t.Error("Expected 'b.d' to be merged") + } + if dst["e"] != "new" { + t.Error("Expected 'e' to be added") + } +} + +func TestMergeDeep_SliceOverwrite(t *testing.T) { + dst := map[string]any{ + "list": []string{"a", "b"}, + } + src := map[string]any{ + "list": []string{"c"}, + } + mergeDeep(dst, src) + + list, ok := dst["list"].([]string) + if !ok { + t.Fatal("Expected list to be []string") + } + if len(list) != 1 || list[0] != "c" { + t.Errorf("Expected list to be overwritten to [c], got %v", list) + } +} + +// --- NeedLeaderElection --- + +func TestNeedLeaderElection(t *testing.T) { + r := &OtelBootstrapRunnable{} + if !r.NeedLeaderElection() { + t.Error("Expected NeedLeaderElection to return true") + } +} + +// --- helpers --- + +func isOpenShift(_ context.Context) (bool, error) { return true, nil } +func notOpenShift(_ context.Context) (bool, error) { return false, nil } +func noMLflowCRD(_ context.Context) (bool, error) { return false, nil } +func mlflowCRDPresent(_ context.Context) (bool, error) { return true, nil } + +func assertContains(t *testing.T, s, substr, msg string) { + t.Helper() + if !contains(s, substr) { + t.Errorf("%s: %q not found in output", msg, substr) + } +} + +func assertNotContains(t *testing.T, s, substr, msg string) { + t.Helper() + if contains(s, substr) { + t.Errorf("%s: %q unexpectedly found in output", msg, substr) + } +} + +func contains(s, substr string) bool { + return len(s) >= len(substr) && searchString(s, substr) +} + +func searchString(s, substr string) bool { + for i := 0; i <= len(s)-len(substr); i++ { + if s[i:i+len(substr)] == substr { + return true + } + } + return false +} diff --git a/kagenti-operator/internal/bootstrap/presets.go b/kagenti-operator/internal/bootstrap/presets.go new file mode 100644 index 00000000..d2ffe252 --- /dev/null +++ b/kagenti-operator/internal/bootstrap/presets.go @@ -0,0 +1,171 @@ +/* +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 bootstrap + +// OTel collector configuration presets ported from the kagenti-deps Helm chart +// (charts/kagenti-deps/values.yaml). The assembleCollectorConfig function merges +// these in the same order as the Helm kagenti.otel.collectorConfig helper. + +const baseConfig = ` +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + http: + endpoint: 0.0.0.0:4318 +exporters: + debug: + verbosity: detailed +processors: + memory_limiter: + check_interval: 1s + limit_mib: 1000 + batch: {} +extensions: + health_check: {} +service: + extensions: [health_check] + pipelines: {} +` + +const defaultPreset = ` +service: + pipelines: + traces/default: + receivers: [otlp] + processors: [memory_limiter, batch] + exporters: [debug] +` + +const phoenixPreset = ` +exporters: + otlp/phoenix: + endpoint: phoenix:4317 + tls: + insecure: true +processors: + filter/phoenix: + traces: + span: + - 'IsMatch(name, "^a2a\\..*")' + - 'attributes["http.method"] != nil' + - 'attributes["mcp.method.name"] == "initialize"' + - 'attributes["mcp.method.name"] == "notifications/initialized"' + - 'attributes["mcp.method.name"] == "notifications/cancelled"' + - 'attributes["mcp.method.name"] == "tools/list"' + transform/genai_to_openinference: + trace_statements: + - context: span + statements: + - >- + set(attributes["llm.model_name"], attributes["gen_ai.request.model"]) + where attributes["gen_ai.request.model"] != nil + - >- + set(attributes["llm.model_name"], attributes["gen_ai.response.model"]) + where attributes["gen_ai.response.model"] != nil and attributes["llm.model_name"] == nil + - >- + set(attributes["llm.token_count.prompt"], attributes["gen_ai.usage.input_tokens"]) + where attributes["gen_ai.usage.input_tokens"] != nil + - >- + set(attributes["llm.token_count.completion"], attributes["gen_ai.usage.output_tokens"]) + where attributes["gen_ai.usage.output_tokens"] != nil + - >- + set(attributes["llm.token_count.total"], + attributes["gen_ai.usage.input_tokens"] + attributes["gen_ai.usage.output_tokens"]) + where attributes["gen_ai.usage.input_tokens"] != nil + and attributes["gen_ai.usage.output_tokens"] != nil + - >- + set(attributes["llm.provider"], attributes["gen_ai.system"]) + where attributes["gen_ai.system"] != nil + - >- + set(attributes["llm.system"], attributes["gen_ai.system"]) + where attributes["gen_ai.system"] != nil + - >- + set(attributes["llm.invocation_parameters"], + Concat(["{\"temperature\":", Concat([attributes["gen_ai.request.temperature"], "}"], "")], "")) + where attributes["gen_ai.request.temperature"] != nil +service: + pipelines: + traces/phoenix: + receivers: [otlp] + processors: [memory_limiter, filter/phoenix, transform/genai_to_openinference, batch] + exporters: [otlp/phoenix] +` + +const mlflowPreset = ` +exporters: + otlphttp/mlflow: + traces_endpoint: http://mlflow:5000/v1/traces + tls: + insecure: true + headers: + x-mlflow-experiment-id: "0" + retry_on_failure: + enabled: true + initial_interval: 5s + max_interval: 30s + max_elapsed_time: 300s + sending_queue: + enabled: true + num_consumers: 2 + queue_size: 1000 +processors: + filter/mlflow: + traces: + span: + - 'IsMatch(name, "^a2a\\..*")' + - 'attributes["http.method"] != nil and IsMatch(name, "^(POST|GET|DELETE)$")' + - 'IsMatch(name, "^mcp-router\\..*") and (attributes["http.method"] == "GET" or attributes["http.method"] == "DELETE")' + - 'attributes["mcp.method.name"] == "initialize"' + - 'attributes["mcp.method.name"] == "notifications/initialized"' + - 'attributes["mcp.method.name"] == "notifications/cancelled"' + - 'attributes["mcp.method.name"] == "tools/list"' +service: + pipelines: + traces/mlflow: + receivers: [otlp] + processors: [memory_limiter, filter/mlflow, batch] + exporters: [debug, otlphttp/mlflow] +` + +const mlflowAuthPreset = ` +extensions: + oauth2client/mlflow: + client_id: ${env:MLFLOW_CLIENT_ID} + client_secret: ${env:MLFLOW_CLIENT_SECRET} + token_url: ${env:KEYCLOAK_TOKEN_URL} + scopes: ["openid"] + timeout: 10s +service: + extensions: [health_check, oauth2client/mlflow] +` + +const rhoaiMlflowAuthPreset = ` +exporters: + otlphttp/mlflow: + 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 +service: + extensions: [health_check, bearertokenauth/mlflow] +` From c889bbc428c9642f91c1f8d676db4c6a9ced0d0a Mon Sep 17 00:00:00 2001 From: Bobbins228 Date: Mon, 25 May 2026 12:56:16 +0100 Subject: [PATCH 2/3] fix: use APIReader for bootstrap and resolve MLflow endpoint from CR The bootstrap runnable was using the manager's cached client which hadn't synced at startup, causing AlreadyExists errors for ConfigMaps created by the Helm chart. Switch all bootstrap reads to APIReader (direct API server) and add AlreadyExists fallback handling. Also fix MLflow endpoint discovery: the MLflow CRD is cluster-scoped so cr.Namespace is always empty. Derive the in-cluster traces endpoint and workspace namespace from status.address.url instead of relying on the hardcoded preset endpoint. Assisted-by: Cursor Signed-off-by: Bobbins228 --- kagenti-operator/internal/bootstrap/otel.go | 162 +++++++++++++----- .../internal/bootstrap/otel_test.go | 6 +- kagenti-operator/internal/mlflow/types.go | 10 ++ 3 files changed, 136 insertions(+), 42 deletions(-) diff --git a/kagenti-operator/internal/bootstrap/otel.go b/kagenti-operator/internal/bootstrap/otel.go index aae93888..a7776e56 100644 --- a/kagenti-operator/internal/bootstrap/otel.go +++ b/kagenti-operator/internal/bootstrap/otel.go @@ -20,6 +20,8 @@ import ( "context" "crypto/sha256" "fmt" + "net/url" + "strings" "time" "github.com/go-logr/logr" @@ -168,7 +170,7 @@ func (r *OtelBootstrapRunnable) reconcileIngressCA(ctx context.Context, log logr existing := &corev1.ConfigMap{} existingKey := types.NamespacedName{Name: ingressCAConfigMap, Namespace: r.Namespace} - if err := r.Client.Get(ctx, existingKey, existing); err != nil { + if err := r.APIReader.Get(ctx, existingKey, existing); err != nil { if !errors.IsNotFound(err) { return fmt.Errorf("checking existing %s ConfigMap: %w", ingressCAConfigMap, err) } @@ -184,10 +186,17 @@ func (r *OtelBootstrapRunnable) reconcileIngressCA(ctx context.Context, log logr Data: map[string]string{caBundleKey: caBundle}, } if err := r.Client.Create(ctx, cm); err != nil { - return fmt.Errorf("creating %s ConfigMap: %w", ingressCAConfigMap, err) + if !errors.IsAlreadyExists(err) { + return fmt.Errorf("creating %s ConfigMap: %w", ingressCAConfigMap, err) + } + log.Info("ConfigMap appeared between check and create, will update", "name", ingressCAConfigMap) + if err := r.APIReader.Get(ctx, existingKey, existing); err != nil { + return fmt.Errorf("re-reading %s ConfigMap: %w", ingressCAConfigMap, err) + } + } else { + log.Info("Created ingress CA ConfigMap", "name", ingressCAConfigMap) + return nil } - log.Info("Created ingress CA ConfigMap", "name", ingressCAConfigMap) - return nil } if existing.Data[caBundleKey] == caBundle { @@ -196,6 +205,11 @@ func (r *OtelBootstrapRunnable) reconcileIngressCA(ctx context.Context, log logr } existing.Data = map[string]string{caBundleKey: caBundle} + if existing.Labels == nil { + existing.Labels = map[string]string{} + } + existing.Labels["app.kubernetes.io/managed-by"] = "kagenti-operator" + existing.Labels["app.kubernetes.io/component"] = "otel-bootstrap" if err := r.Client.Update(ctx, existing); err != nil { return fmt.Errorf("updating %s ConfigMap: %w", ingressCAConfigMap, err) } @@ -206,14 +220,14 @@ func (r *OtelBootstrapRunnable) reconcileIngressCA(ctx context.Context, log logr // reconcileCollectorConfig discovers available components and assembles the // OTel collector ConfigMap from preset configurations. func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, log logr.Logger, isOCP bool) error { - mlflowAvailable, mlflowNamespace, err := r.discoverMLflow(ctx, log) + mf, err := r.discoverMLflow(ctx, log) if err != nil { return err } phoenixAvailable := r.discoverPhoenix(ctx, log) - config, err := assembleCollectorConfig(isOCP, mlflowAvailable, mlflowNamespace, phoenixAvailable) + config, err := assembleCollectorConfig(isOCP, mf, phoenixAvailable) if err != nil { return fmt.Errorf("assembling collector config: %w", err) } @@ -228,7 +242,7 @@ func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, lo existing := &corev1.ConfigMap{} key := types.NamespacedName{Name: collectorConfigMapName, Namespace: r.Namespace} - if err := r.Client.Get(ctx, key, existing); err != nil { + if err := r.APIReader.Get(ctx, key, existing); err != nil { if !errors.IsNotFound(err) { return fmt.Errorf("checking existing collector ConfigMap: %w", err) } @@ -247,11 +261,18 @@ func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, lo Data: map[string]string{configMapDataKey: configStr}, } if err := r.Client.Create(ctx, cm); err != nil { - return fmt.Errorf("creating collector ConfigMap: %w", err) + if !errors.IsAlreadyExists(err) { + return fmt.Errorf("creating collector ConfigMap: %w", err) + } + log.Info("ConfigMap appeared between check and create, will update", "name", collectorConfigMapName) + if err := r.APIReader.Get(ctx, key, existing); err != nil { + return fmt.Errorf("re-reading collector ConfigMap: %w", err) + } + } else { + log.Info("Created OTel collector ConfigMap", "components", + componentSummary(mf.available, phoenixAvailable)) + return r.rolloutRestartCollector(ctx, log, configHash) } - log.Info("Created OTel collector ConfigMap", "components", - componentSummary(mlflowAvailable, phoenixAvailable)) - return r.rolloutRestartCollector(ctx, log, configHash) } existingHash := existing.Annotations[restartAnnotation] @@ -265,60 +286,105 @@ func (r *OtelBootstrapRunnable) reconcileCollectorConfig(ctx context.Context, lo existing.Annotations = make(map[string]string) } existing.Annotations[restartAnnotation] = configHash + if existing.Labels == nil { + existing.Labels = map[string]string{} + } + existing.Labels["app.kubernetes.io/managed-by"] = "kagenti-operator" + existing.Labels["app.kubernetes.io/component"] = "otel-bootstrap" if err := r.Client.Update(ctx, existing); err != nil { return fmt.Errorf("updating collector ConfigMap: %w", err) } log.Info("Updated OTel collector ConfigMap", "components", - componentSummary(mlflowAvailable, phoenixAvailable)) + componentSummary(mf.available, phoenixAvailable)) return r.rolloutRestartCollector(ctx, log, configHash) } +// mlflowInfo holds the discovered MLflow endpoint and workspace namespace. +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 +} + // discoverMLflow checks for the MLflow CRD and, if present, discovers the -// MLflow CR to derive the namespace where the service runs. -func (r *OtelBootstrapRunnable) discoverMLflow(ctx context.Context, log logr.Logger) (available bool, namespace string, err error) { +// MLflow CR to derive the in-cluster endpoint and workspace namespace. +func (r *OtelBootstrapRunnable) discoverMLflow(ctx context.Context, log logr.Logger) (*mlflowInfo, error) { crdExists, err := r.mlflowCRDPresent(ctx) if err != nil { - return false, "", fmt.Errorf("checking MLflow CRD: %w", err) + return nil, fmt.Errorf("checking MLflow CRD: %w", err) } if !crdExists { log.Info("MLflow CRD (mlflows.mlflow.opendatahub.io) not found. " + "MLflow presets will be skipped. If the MLflow operator is installed later, " + "restart the kagenti-operator pod to pick up MLflow configuration.") - return false, "", nil + return &mlflowInfo{}, nil } log.Info("MLflow CRD detected, discovering MLflow CR") list := &mlflow.MLflowList{} - if err := r.Client.List(ctx, list); err != nil { + if err := r.APIReader.List(ctx, list); err != nil { log.Info("Could not list MLflow CRs, skipping MLflow presets", "error", err) - return false, "", nil + return &mlflowInfo{}, nil } for i := range list.Items { cr := &list.Items[i] if meta.IsStatusConditionTrue(cr.Status.Conditions, "Available") { - log.Info("Found available MLflow CR", - "name", cr.Name, "namespace", cr.Namespace, "url", cr.Status.URL) - return true, cr.Namespace, nil + info := mlflowInfoFromCR(cr, log) + return info, nil } } log.Info("MLflow CRD present but no Available MLflow CR found, waiting for service readiness") - available, namespace, err = r.waitForMLflowService(ctx, log) + info, err := r.waitForMLflowService(ctx, log) if err != nil { log.Info("MLflow service did not become ready within timeout, skipping MLflow presets", "error", err) - return false, "", nil + return &mlflowInfo{}, nil } - return available, namespace, nil + 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"). +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 { + 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) + return info + } + log.Info("Could not parse MLflow address URL, falling back", "url", cr.Status.Address.URL, "error", err) + } + + if cr.Status.URL != "" { + info.tracesURL = strings.TrimRight(cr.Status.URL, "/") + "/v1/traces" + log.Info("Found available MLflow CR (using external URL)", + "name", cr.Name, "url", cr.Status.URL, "tracesURL", info.tracesURL) + } else { + log.Info("Found available MLflow CR but no endpoint URL in status", + "name", cr.Name) + } + return info } // waitForMLflowService retries with backoff until an MLflow CR becomes Available. -func (r *OtelBootstrapRunnable) waitForMLflowService(ctx context.Context, log logr.Logger) (bool, string, error) { - var resultAvailable bool - var resultNamespace string +func (r *OtelBootstrapRunnable) waitForMLflowService(ctx context.Context, log logr.Logger) (*mlflowInfo, error) { + var result *mlflowInfo backoff := wait.Backoff{ Duration: defaultBackoffInitial, @@ -332,17 +398,14 @@ func (r *OtelBootstrapRunnable) waitForMLflowService(ctx context.Context, log lo err := wait.ExponentialBackoffWithContext(timeoutCtx, backoff, func(ctx context.Context) (bool, error) { list := &mlflow.MLflowList{} - if err := r.Client.List(ctx, list); err != nil { + if err := r.APIReader.List(ctx, list); err != nil { log.V(1).Info("Retrying MLflow CR list", "error", err) return false, nil } for i := range list.Items { cr := &list.Items[i] if meta.IsStatusConditionTrue(cr.Status.Conditions, "Available") { - log.Info("MLflow CR became Available", - "name", cr.Name, "namespace", cr.Namespace) - resultAvailable = true - resultNamespace = cr.Namespace + result = mlflowInfoFromCR(cr, log) return true, nil } } @@ -350,7 +413,10 @@ func (r *OtelBootstrapRunnable) waitForMLflowService(ctx context.Context, log lo return false, nil }) - return resultAvailable, resultNamespace, err + if result == nil { + result = &mlflowInfo{} + } + return result, err } // mlflowCRDPresent checks if the mlflows.mlflow.opendatahub.io CRD is installed. @@ -380,7 +446,7 @@ func (r *OtelBootstrapRunnable) mlflowCRDPresent(ctx context.Context) (bool, err func (r *OtelBootstrapRunnable) discoverPhoenix(ctx context.Context, log logr.Logger) bool { svc := &corev1.Service{} key := types.NamespacedName{Name: phoenixServiceName, Namespace: r.Namespace} - if err := r.Client.Get(ctx, key, svc); err != nil { + if err := r.APIReader.Get(ctx, key, svc); err != nil { log.V(1).Info("Phoenix service not found, skipping Phoenix preset", "namespace", r.Namespace) return false @@ -394,7 +460,7 @@ func (r *OtelBootstrapRunnable) discoverPhoenix(ctx context.Context, log logr.Lo func (r *OtelBootstrapRunnable) rolloutRestartCollector(ctx context.Context, log logr.Logger, configHash string) error { dep := &appsv1.Deployment{} key := types.NamespacedName{Name: collectorDeployment, Namespace: r.Namespace} - if err := r.Client.Get(ctx, key, dep); err != nil { + if err := r.APIReader.Get(ctx, key, dep); err != nil { if errors.IsNotFound(err) { log.Info("OTel collector Deployment not found, skipping rollout restart") return nil @@ -421,7 +487,7 @@ func (r *OtelBootstrapRunnable) rolloutRestartCollector(ctx context.Context, log // assembleCollectorConfig builds the complete OTel collector YAML config by // merging the base config with component-specific presets. -func assembleCollectorConfig(isOCP, mlflowAvailable bool, mlflowNamespace string, phoenixAvailable bool) (map[string]any, error) { +func assembleCollectorConfig(isOCP bool, mf *mlflowInfo, phoenixAvailable bool) (map[string]any, error) { config, err := parsePreset(baseConfig) if err != nil { return nil, fmt.Errorf("parsing base config: %w", err) @@ -438,7 +504,7 @@ func assembleCollectorConfig(isOCP, mlflowAvailable bool, mlflowNamespace string hasComponentPipeline = true } - if mlflowAvailable { + if mf.available { mlflowCfg, err := parsePreset(mlflowPreset) if err != nil { return nil, fmt.Errorf("parsing mlflow preset: %w", err) @@ -446,6 +512,10 @@ func assembleCollectorConfig(isOCP, mlflowAvailable bool, mlflowNamespace string mergeDeep(config, mlflowCfg) hasComponentPipeline = true + if mf.tracesURL != "" { + setMLflowTracesEndpoint(config, mf.tracesURL) + } + if isOCP { rhoaiAuth, err := parsePreset(rhoaiMlflowAuthPreset) if err != nil { @@ -454,7 +524,7 @@ func assembleCollectorConfig(isOCP, mlflowAvailable bool, mlflowNamespace string mergeDeep(config, rhoaiAuth) clearMLflowExporterTLS(config) - setMLflowBearerTokenAuth(config, mlflowNamespace) + setMLflowBearerTokenAuth(config, mf.workspaceNS) } else { mlflowAuth, err := parsePreset(mlflowAuthPreset) if err != nil { @@ -474,13 +544,27 @@ func assembleCollectorConfig(isOCP, mlflowAvailable bool, mlflowNamespace string mergeDeep(config, defaultCfg) } - if isOCP && mlflowAvailable { + if isOCP && mf.available { setIngressCATLS(config) } return config, nil } +// setMLflowTracesEndpoint overrides the hardcoded preset endpoint with the +// dynamically discovered in-cluster URL from the MLflow CR. +func setMLflowTracesEndpoint(config map[string]any, tracesURL string) { + exporters, ok := config["exporters"].(map[string]any) + if !ok { + return + } + mlflowExp, ok := exporters["otlphttp/mlflow"].(map[string]any) + if !ok { + return + } + mlflowExp["traces_endpoint"] = tracesURL +} + // parsePreset unmarshals a YAML string into a map. func parsePreset(yamlStr string) (map[string]any, error) { var result map[string]any diff --git a/kagenti-operator/internal/bootstrap/otel_test.go b/kagenti-operator/internal/bootstrap/otel_test.go index 640c2c1e..2ab145bd 100644 --- a/kagenti-operator/internal/bootstrap/otel_test.go +++ b/kagenti-operator/internal/bootstrap/otel_test.go @@ -483,7 +483,7 @@ func TestConfigMapChanged_TriggersRestart(t *testing.T) { // --- assembleCollectorConfig unit tests --- func TestAssembleConfig_DefaultOnly(t *testing.T) { - cfg, err := assembleCollectorConfig(false, false, "", false) + cfg, err := assembleCollectorConfig(false, &mlflowInfo{}, false) if err != nil { t.Fatalf("assembleCollectorConfig failed: %v", err) } @@ -502,7 +502,7 @@ func TestAssembleConfig_DefaultOnly(t *testing.T) { } func TestAssembleConfig_PhoenixAndMLflow(t *testing.T) { - cfg, err := assembleCollectorConfig(false, true, "mlflow-ns", true) + cfg, err := assembleCollectorConfig(false, &mlflowInfo{available: true, tracesURL: "http://mlflow.mlflow-ns.svc:5000/v1/traces", workspaceNS: "mlflow-ns"}, true) if err != nil { t.Fatalf("assembleCollectorConfig failed: %v", err) } @@ -522,7 +522,7 @@ func TestAssembleConfig_PhoenixAndMLflow(t *testing.T) { } func TestAssembleConfig_OCP_MLflow_IngressCATLS(t *testing.T) { - cfg, err := assembleCollectorConfig(true, true, "rhoai-ns", false) + cfg, err := assembleCollectorConfig(true, &mlflowInfo{available: true, tracesURL: "https://mlflow.rhoai-ns.svc:8443/v1/traces", workspaceNS: "rhoai-ns"}, false) if err != nil { t.Fatalf("assembleCollectorConfig failed: %v", err) } diff --git a/kagenti-operator/internal/mlflow/types.go b/kagenti-operator/internal/mlflow/types.go index 1649bae1..9abb5df5 100644 --- a/kagenti-operator/internal/mlflow/types.go +++ b/kagenti-operator/internal/mlflow/types.go @@ -54,6 +54,13 @@ 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). URL string `json:"url,omitempty"` + // Address holds the in-cluster address for the MLflow server. + Address *MLflowAddress `json:"address,omitempty"` +} + +type MLflowAddress struct { + // URL is the in-cluster service URL (e.g. https://mlflow.redhat-ods-applications.svc:8443). + URL string `json:"url,omitempty"` } type MLflowList struct { @@ -90,6 +97,9 @@ func (in *MLflowStatus) DeepCopyInto(out *MLflowStatus) { in.Conditions[i].DeepCopyInto(&out.Conditions[i]) } } + if in.Address != nil { + out.Address = &MLflowAddress{URL: in.Address.URL} + } } func (in *MLflowList) DeepCopyObject() runtime.Object { From 44016d4cb372d0e257ed4f46fef85b0cb9e0b9e1 Mon Sep 17 00:00:00 2001 From: Bobbins228 Date: Tue, 26 May 2026 16:21:20 +0100 Subject: [PATCH 3/3] fix: guard mlflowInfoFromCR against malformed URLs and add tests Guard against empty scheme/host when parsing MLflow address URL, falling back to Status.URL. Add three negative test cases for mlflowInfoFromCR exercising empty, malformed, and scheme-less URLs. Assisted-By: Cursor Signed-off-by: Bobbins228 --- kagenti-operator/internal/bootstrap/otel.go | 5 +- .../internal/bootstrap/otel_test.go | 53 +++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/kagenti-operator/internal/bootstrap/otel.go b/kagenti-operator/internal/bootstrap/otel.go index a7776e56..8f3017f0 100644 --- a/kagenti-operator/internal/bootstrap/otel.go +++ b/kagenti-operator/internal/bootstrap/otel.go @@ -356,7 +356,7 @@ func mlflowInfoFromCR(cr *mlflow.MLflow, log logr.Logger) *mlflowInfo { if cr.Status.Address != nil && cr.Status.Address.URL != "" { parsed, err := url.Parse(cr.Status.Address.URL) - if err == nil { + if err == nil && parsed.Scheme != "" && parsed.Host != "" { hostname := parsed.Hostname() parts := strings.SplitN(hostname, ".", 3) if len(parts) >= 2 { @@ -368,7 +368,8 @@ func mlflowInfoFromCR(cr *mlflow.MLflow, log logr.Logger) *mlflowInfo { "tracesURL", info.tracesURL, "workspaceNS", info.workspaceNS) return info } - log.Info("Could not parse MLflow address URL, falling back", "url", cr.Status.Address.URL, "error", err) + log.Info("MLflow address URL missing scheme or host, falling back", + "url", cr.Status.Address.URL) } if cr.Status.URL != "" { diff --git a/kagenti-operator/internal/bootstrap/otel_test.go b/kagenti-operator/internal/bootstrap/otel_test.go index 2ab145bd..6d5a6556 100644 --- a/kagenti-operator/internal/bootstrap/otel_test.go +++ b/kagenti-operator/internal/bootstrap/otel_test.go @@ -329,6 +329,59 @@ func TestMLflowCRDPresent_NonOCP_UsesOAuthAuth(t *testing.T) { assertContains(t, config, "oauth2client/mlflow", "Expected OAuth2 client auth on non-OCP") } +// --- mlflowInfoFromCR negative tests --- + +func TestMLflowInfoFromCR_EmptyAddressURL(t *testing.T) { + cr := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{Name: "mlflow"}, + Status: mlflow.MLflowStatus{ + Address: &mlflow.MLflowAddress{URL: ""}, + URL: "https://mlflow-external.example.com", + }, + } + info := mlflowInfoFromCR(cr, logr.Discard()) + if !info.available { + t.Fatal("Expected available=true") + } + if info.tracesURL != "https://mlflow-external.example.com/v1/traces" { + t.Fatalf("Expected fallback to Status.URL, got tracesURL=%q", info.tracesURL) + } +} + +func TestMLflowInfoFromCR_MalformedURL(t *testing.T) { + cr := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{Name: "mlflow"}, + Status: mlflow.MLflowStatus{ + Address: &mlflow.MLflowAddress{URL: "://badurl"}, + URL: "https://mlflow-fallback.example.com", + }, + } + info := mlflowInfoFromCR(cr, logr.Discard()) + if !info.available { + t.Fatal("Expected available=true") + } + if info.tracesURL != "https://mlflow-fallback.example.com/v1/traces" { + t.Fatalf("Expected fallback to Status.URL for malformed address, got tracesURL=%q", info.tracesURL) + } +} + +func TestMLflowInfoFromCR_MissingSchemeFallback(t *testing.T) { + cr := &mlflow.MLflow{ + ObjectMeta: metav1.ObjectMeta{Name: "mlflow"}, + Status: mlflow.MLflowStatus{ + Address: &mlflow.MLflowAddress{URL: "//mlflow.ns.svc:8443"}, + URL: "https://mlflow-fallback.example.com", + }, + } + info := mlflowInfoFromCR(cr, logr.Discard()) + if !info.available { + t.Fatal("Expected available=true") + } + if info.tracesURL != "https://mlflow-fallback.example.com/v1/traces" { + t.Fatalf("Expected fallback when scheme is empty, got tracesURL=%q", info.tracesURL) + } +} + // --- Phoenix tests --- func TestPhoenixPresent_MergesPhoenixPreset(t *testing.T) {