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..8f3017f0 --- /dev/null +++ b/kagenti-operator/internal/bootstrap/otel.go @@ -0,0 +1,684 @@ +/* +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" + "net/url" + "strings" + "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.APIReader.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 { + 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 + } + } + + 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 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) + } + 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 { + mf, err := r.discoverMLflow(ctx, log) + if err != nil { + return err + } + + phoenixAvailable := r.discoverPhoenix(ctx, log) + + config, err := assembleCollectorConfig(isOCP, mf, 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.APIReader.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 { + 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) + } + } + + 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 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(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 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 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 &mlflowInfo{}, nil + } + + log.Info("MLflow CRD detected, discovering MLflow CR") + + list := &mlflow.MLflowList{} + if err := r.APIReader.List(ctx, list); err != nil { + log.Info("Could not list MLflow CRs, skipping MLflow presets", "error", err) + return &mlflowInfo{}, nil + } + + for i := range list.Items { + cr := &list.Items[i] + if meta.IsStatusConditionTrue(cr.Status.Conditions, "Available") { + info := mlflowInfoFromCR(cr, log) + return info, nil + } + } + + log.Info("MLflow CRD present but no Available MLflow CR found, waiting for service readiness") + 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 &mlflowInfo{}, 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 && 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) + return info + } + log.Info("MLflow address URL missing scheme or host, falling back", + "url", cr.Status.Address.URL) + } + + 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) (*mlflowInfo, error) { + var result *mlflowInfo + + 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.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") { + result = mlflowInfoFromCR(cr, log) + return true, nil + } + } + log.V(1).Info("MLflow CR not yet Available, retrying") + return false, nil + }) + + if result == nil { + result = &mlflowInfo{} + } + return result, 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.APIReader.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.APIReader.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 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) + } + + 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 mf.available { + mlflowCfg, err := parsePreset(mlflowPreset) + if err != nil { + return nil, fmt.Errorf("parsing mlflow preset: %w", err) + } + mergeDeep(config, mlflowCfg) + hasComponentPipeline = true + + if mf.tracesURL != "" { + setMLflowTracesEndpoint(config, mf.tracesURL) + } + + 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, mf.workspaceNS) + } 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 && 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 + 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..6d5a6556 --- /dev/null +++ b/kagenti-operator/internal/bootstrap/otel_test.go @@ -0,0 +1,699 @@ +/* +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") +} + +// --- 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) { + 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, &mlflowInfo{}, 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, &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) + } + + 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, &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) + } + + 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] +` 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 {