diff --git a/bindata/observability/07-observability-operator.yaml b/bindata/observability/07-observability-operator.yaml new file mode 100644 index 0000000000..9fa7604b84 --- /dev/null +++ b/bindata/observability/07-observability-operator.yaml @@ -0,0 +1,252 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: netobserv-operator +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: netobserv-operator-installer + namespace: netobserv-operator +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: netobserv-operator-installer +rules: + # Permissions needed by OLM v1 to install the Network Observability operator + # Based on OLM v1 pre-authorization requirements from the operator bundle + + # Core resources + - apiGroups: [""] + resources: ["endpoints", "nodes", "pods"] + verbs: ["get", "list", "watch"] + - apiGroups: [""] + resources: ["events"] + verbs: ["create", "patch"] + - apiGroups: [""] + resources: ["namespaces"] + verbs: ["create", "get", "list", "watch", "update", "patch", "delete"] + - apiGroups: [""] + resources: ["serviceaccounts", "services", "configmaps"] + verbs: ["create", "get", "list", "watch", "update", "patch", "delete"] + - apiGroups: [""] + resources: ["secrets", "persistentvolumeclaims"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + + # Webhooks - validating webhook configurations + - apiGroups: ["admissionregistration.k8s.io"] + resources: ["validatingwebhookconfigurations"] + verbs: ["create", "list", "watch"] + - apiGroups: ["admissionregistration.k8s.io"] + resources: ["validatingwebhookconfigurations"] + resourceNames: ["flowcollectorconversionwebhook.netobserv.io", "flowmetricvalidationwebhook.netobserv.io"] + verbs: ["delete", "get", "patch", "update"] + + # CRDs and API services + - apiGroups: ["apiextensions.k8s.io"] + resources: ["customresourcedefinitions"] + verbs: ["create", "get", "list", "watch", "update", "patch"] + - apiGroups: ["apiextensions.k8s.io"] + resources: ["customresourcedefinitions/status"] + verbs: ["patch", "update"] + - apiGroups: ["apiextensions.k8s.io"] + resources: ["customresourcedefinitions"] + resourceNames: ["flowcollectors.flows.netobserv.io", "flowcollectorslices.flows.netobserv.io", "flowmetrics.flows.netobserv.io"] + verbs: ["delete"] + - apiGroups: ["apiregistration.k8s.io"] + resources: ["apiservices"] + verbs: ["get", "list", "watch"] + + # Workloads + - apiGroups: ["apps"] + resources: ["deployments", "daemonsets"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + - apiGroups: ["apps"] + resources: ["replicasets"] + verbs: ["get", "list", "watch"] + + # Authentication and authorization + - apiGroups: ["authentication.k8s.io"] + resources: ["tokenreviews"] + verbs: ["create"] + - apiGroups: ["authorization.k8s.io"] + resources: ["subjectaccessreviews"] + verbs: ["create"] + + # Autoscaling + - apiGroups: ["autoscaling"] + resources: ["horizontalpodautoscalers"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + + # eBPF (bpfman.io) + - apiGroups: ["bpfman.io"] + resources: ["clusterbpfapplications"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + - apiGroups: ["bpfman.io"] + resources: ["clusterbpfapplications/status"] + verbs: ["get", "patch", "update"] + + # OpenShift config + - apiGroups: ["config.openshift.io"] + resources: ["clusterversions", "networks"] + verbs: ["get", "list", "watch"] + + # Console plugin + - apiGroups: ["console.openshift.io"] + resources: ["consoleplugins"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + + # Coordination (leader election) + - apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + + # Discovery + - apiGroups: ["discovery.k8s.io"] + resources: ["endpointslices"] + verbs: ["get", "list", "watch"] + + # FlowCollector CRs + - apiGroups: ["flows.netobserv.io"] + resources: ["flowcollectors", "flowcollectorslices", "flowmetrics"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + - apiGroups: ["flows.netobserv.io"] + resources: ["flowcollectors/finalizers"] + verbs: ["update"] + - apiGroups: ["flows.netobserv.io"] + resources: ["flowcollectors/status", "flowcollectorslices/status", "flowmetrics/status"] + verbs: ["get", "patch", "update"] + + # OVN-Kubernetes + - apiGroups: ["k8s.ovn.org"] + resources: ["clusteruserdefinednetworks", "userdefinednetworks"] + verbs: ["get", "list", "watch"] + + # Loki integration + - apiGroups: ["loki.grafana.com"] + resources: ["lokistacks"] + verbs: ["get", "list", "watch"] + - apiGroups: ["loki.grafana.com"] + resources: ["network"] + resourceNames: ["logs"] + verbs: ["create", "get"] + + # Metrics + - apiGroups: ["metrics.k8s.io"] + resources: ["pods"] + verbs: ["create"] + + # Monitoring + - apiGroups: ["monitoring.coreos.com"] + resources: ["prometheusrules", "servicemonitors"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + + # Network policies + - apiGroups: ["networking.k8s.io"] + resources: ["networkpolicies"] + verbs: ["create", "delete", "get", "list", "patch", "update", "watch"] + + # OLM ClusterExtension finalizers + - apiGroups: ["olm.operatorframework.io"] + resources: ["clusterextensions/finalizers"] + resourceNames: ["netobserv-operator"] + verbs: ["update"] + + # OLM ClusterObjectSet finalizers + - apiGroups: ["olm.operatorframework.io"] + resources: ["clusterobjectsets/finalizers"] + resourceNames: ["netobserv-operator-1"] + verbs: ["update"] + + # OpenShift console and network config + - apiGroups: ["operator.openshift.io"] + resources: ["consoles"] + verbs: ["get", "list", "update", "watch"] + - apiGroups: ["operator.openshift.io"] + resources: ["networks"] + verbs: ["get", "list", "watch"] + + # RBAC + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["clusterroles", "clusterrolebindings"] + verbs: ["create", "delete", "get", "list", "update", "watch", "patch"] + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["roles"] + verbs: ["list", "watch"] + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["rolebindings"] + verbs: ["create", "delete", "get", "list", "update", "watch"] + + # Security context constraints + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + verbs: ["create", "list", "update", "watch"] + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + resourceNames: ["hostnetwork"] + verbs: ["use"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: netobserv-operator-installer +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: netobserv-operator-installer +subjects: + - kind: ServiceAccount + name: netobserv-operator-installer + namespace: netobserv-operator +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: netobserv-operator-installer + namespace: netobserv-operator +rules: + # Namespace-scoped permissions needed for operator installation + # The operator needs to manage Roles and RoleBindings in its own namespace + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["roles"] + verbs: ["create"] + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["roles"] + resourceNames: ["netobserv-openshift-netobserv-operator-prometheus"] + verbs: ["delete", "get", "patch", "update"] + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["rolebindings"] + resourceNames: ["netobserv-openshift-netobserv-operator-prometheus"] + verbs: ["delete", "get", "patch", "update"] + - apiGroups: ["rbac.authorization.k8s.io"] + resources: ["rolebindings"] + verbs: ["create"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: netobserv-operator-installer + namespace: netobserv-operator +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: netobserv-operator-installer +subjects: + - kind: ServiceAccount + name: netobserv-operator-installer + namespace: netobserv-operator +--- +apiVersion: olm.operatorframework.io/v1 +kind: ClusterExtension +metadata: + name: netobserv-operator +spec: + namespace: netobserv-operator + serviceAccount: + name: netobserv-operator-installer + source: + sourceType: Catalog + catalog: + packageName: netobserv-operator + channels: [stable] diff --git a/bindata/observability/08-flowcollector.yaml b/bindata/observability/08-flowcollector.yaml new file mode 100644 index 0000000000..21b1062493 --- /dev/null +++ b/bindata/observability/08-flowcollector.yaml @@ -0,0 +1,25 @@ +apiVersion: flows.netobserv.io/v1beta2 +kind: FlowCollector +metadata: + name: cluster +spec: + agent: + ebpf: + features: + - DNSTracking + sampling: 400 + metrics: + server: + tls: + type: Auto + type: eBPF + processor: + metrics: + server: + tls: + type: Auto + + deploymentModel: Service + loki: + enable: false + namespace: netobserv diff --git a/pkg/controller/add_networkconfig.go b/pkg/controller/add_networkconfig.go index 0c5f6bceb3..2952cad174 100644 --- a/pkg/controller/add_networkconfig.go +++ b/pkg/controller/add_networkconfig.go @@ -8,6 +8,7 @@ import ( "github.com/openshift/cluster-network-operator/pkg/controller/egress_router" "github.com/openshift/cluster-network-operator/pkg/controller/infrastructureconfig" "github.com/openshift/cluster-network-operator/pkg/controller/ingressconfig" + "github.com/openshift/cluster-network-operator/pkg/controller/observability" "github.com/openshift/cluster-network-operator/pkg/controller/operconfig" pkictrl "github.com/openshift/cluster-network-operator/pkg/controller/pki" "github.com/openshift/cluster-network-operator/pkg/controller/proxyconfig" @@ -27,5 +28,6 @@ func init() { allowlist.Add, dashboards.Add, pkictrl.Add, + observability.Add, ) } diff --git a/pkg/controller/observability/observability_controller.go b/pkg/controller/observability/observability_controller.go new file mode 100644 index 0000000000..edd746a1a9 --- /dev/null +++ b/pkg/controller/observability/observability_controller.go @@ -0,0 +1,561 @@ +package observability + +import ( + "bytes" + "context" + "fmt" + "io" + "os" + "strings" + "time" + + configv1 "github.com/openshift/api/config/v1" + operatorv1 "github.com/openshift/api/operator/v1" + cnoclient "github.com/openshift/cluster-network-operator/pkg/client" + "github.com/openshift/cluster-network-operator/pkg/controller/statusmanager" + "github.com/openshift/library-go/pkg/operator/configobserver/featuregates" + operatorv1helpers "github.com/openshift/library-go/pkg/operator/v1helpers" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/yaml" + "k8s.io/klog/v2" + crclient "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/controller-runtime/pkg/source" +) + +const ( + OperatorYAML = "bindata/observability/07-observability-operator.yaml" + FlowCollectorYAML = "bindata/observability/08-flowcollector.yaml" + NetObservNamespace = "netobserv" + OperatorNamespace = "netobserv-operator" + FlowCollectorVersion = "v1beta2" + FlowCollectorName = "cluster" + NetworkCRName = "cluster" + + NetworkObservabilityDeployed = "NetworkObservabilityDeployed" + + checkInterval = 10 * time.Second + checkTimeout = 10 * time.Minute + requeueAfterOLM = 5 * time.Minute // Requeue interval for OLM operations (install/wait) + requeueAfterStandard = 30 * time.Second // Requeue interval for standard operations +) + +// Add creates a new controller. Referenced in add_networkconfig.go. +func Add(mgr manager.Manager, _ *statusmanager.StatusManager, _ cnoclient.Client, featureGate featuregates.FeatureGate) error { + klog.Info("Add Network Observability Operator to manager") + return add(mgr, newReconciler(mgr.GetClient(), featureGate)) +} + +func newReconciler(client crclient.Client, featureGate featuregates.FeatureGate) *ReconcileObservability { + return &ReconcileObservability{ + client: client, + featureGate: featureGate, + } +} + +func add(mgr manager.Manager, r *ReconcileObservability) error { + c, err := controller.New("observability-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + + // Watch config.openshift.io/v1 Network CR for spec changes + if err := c.Watch(source.Kind(mgr.GetCache(), &configv1.Network{}, &handler.TypedEnqueueRequestForObject[*configv1.Network]{})); err != nil { + return err + } + + // Watch operator.openshift.io/v1 Network CR for status changes + // This ensures that status condition changes trigger reconciliation + return c.Watch(source.Kind(mgr.GetCache(), &operatorv1.Network{}, &handler.TypedEnqueueRequestForObject[*operatorv1.Network]{})) +} + +var _ reconcile.Reconciler = &ReconcileObservability{} + +type ReconcileObservability struct { + client crclient.Client + featureGate featuregates.FeatureGate +} + +// Reconcile reacts to changes in Network CR +func (r *ReconcileObservability) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + klog.Info("Reconcile Network Observability") + + if req.Name != NetworkCRName { + return reconcile.Result{}, nil // only reconcile the singleton Network object + } + + // Check if NetworkObservabilityInstall feature gate is enabled + if !r.isFeatureGateEnabled() { + klog.V(4).Info("NetworkObservabilityInstall feature gate is disabled, skipping Network Observability management") + return reconcile.Result{}, nil + } + + // Check if Network Observability should be enabled + shouldInstall, err := r.shouldInstallNetworkObservability(ctx) + if err != nil { + klog.Warningf("Failed to determine if Network Observability should be installed: %v. Will retry in %v.", err, requeueAfterStandard) + return reconcile.Result{RequeueAfter: requeueAfterStandard}, nil + } + if !shouldInstall { + return reconcile.Result{}, nil + } + + // Proceed with installation/reinstallation + installed, ceExists, err := r.isNetObservOperatorInstalled(ctx) + if err != nil { + klog.Warningf("Failed to check if Network Observability Operator is installed: %v. Will retry in %v.", err, requeueAfterStandard) + // Mark deployment as failed with the error details + _ = r.setNetworkObservabilityCondition(ctx, operatorv1.ConditionFalse, "DeploymentFailed", fmt.Sprintf("Failed to check Network Observability Operator status: %v", err)) + return reconcile.Result{RequeueAfter: requeueAfterStandard}, nil + } + if !installed { + if !ceExists { + // ClusterExtension doesn't exist yet, create it + if err := r.installNetObservOperator(ctx); err != nil { + klog.Warningf("Failed to install Network Observability Operator: %v. Will retry in %v.", err, requeueAfterOLM) + _ = r.setNetworkObservabilityCondition(ctx, operatorv1.ConditionFalse, "DeploymentFailed", fmt.Sprintf("Failed to install Network Observability Operator: %v", err)) + return reconcile.Result{RequeueAfter: requeueAfterOLM}, nil + } + klog.Infof("Created ClusterExtension netobserv-operator, will check installation status in %v", requeueAfterOLM) + return reconcile.Result{RequeueAfter: requeueAfterOLM}, nil + } + + // ClusterExtension exists but installation not complete yet + klog.Infof("ClusterExtension netobserv-operator installation in progress, will recheck in %v", requeueAfterOLM) + return reconcile.Result{RequeueAfter: requeueAfterOLM}, nil + } + + // Operator installation completed + klog.Info("ClusterExtension netobserv-operator installation completed, proceeding to FlowCollector creation") + + // Check if FlowCollector already exists + flowCollectorExists, err := r.isFlowCollectorExists(ctx) + if err != nil { + klog.Warningf("Failed to check if FlowCollector exists: %v. Will retry in %v.", err, requeueAfterStandard) + return reconcile.Result{RequeueAfter: requeueAfterStandard}, nil + } + + if !flowCollectorExists { + // Create FlowCollector + if err := r.createFlowCollector(ctx); err != nil { + klog.Warningf("Failed to create FlowCollector: %v. Will retry in %v.", err, requeueAfterStandard) + // Mark deployment as failed + _ = r.setNetworkObservabilityCondition(ctx, operatorv1.ConditionFalse, "DeploymentFailed", fmt.Sprintf("Failed to create FlowCollector: %v", err)) + return reconcile.Result{RequeueAfter: requeueAfterStandard}, nil + } + klog.Info("FlowCollector created successfully") + } + + // Mark as deployed to track deployment status + if err := r.setNetworkObservabilityCondition(ctx, operatorv1.ConditionTrue, "DeploymentComplete", "Network Observability has been deployed"); err != nil { + klog.Warningf("Failed to mark Network Observability as deployed: %v. Will retry in %v.", err, requeueAfterStandard) + return reconcile.Result{RequeueAfter: requeueAfterStandard}, nil + } + + klog.V(4).Info("Network Observability is deployed") + return reconcile.Result{}, nil +} + +// isFeatureGateEnabled checks if the NetworkObservabilityInstall feature gate is enabled. +// If featureGate is nil (e.g., in tests), returns false to default to disabled. +// If the feature gate is not registered yet (older cluster versions), returns false. +func (r *ReconcileObservability) isFeatureGateEnabled() bool { + if r.featureGate == nil { + return false // Default to disabled in tests + } + + featureGateName := configv1.FeatureGateName("NetworkObservabilityInstall") + + return r.featureGate.Enabled(featureGateName) +} + +// wasNetworkObservabilityDeployed checks if the NetworkObservabilityDeployed condition +// is set to True in the network.operator.openshift.io Network CR status +func (r *ReconcileObservability) wasNetworkObservabilityDeployed(ctx context.Context) (bool, error) { + network := &operatorv1.Network{} + if err := r.client.Get(ctx, types.NamespacedName{Name: NetworkCRName}, network); err != nil { + return false, err + } + + for _, condition := range network.Status.Conditions { + if condition.Type == NetworkObservabilityDeployed { + return condition.Status == operatorv1.ConditionTrue, nil + } + } + + return false, nil +} + +// setNetworkObservabilityCondition sets the NetworkObservabilityDeployed condition +// in the network.operator.openshift.io Network CR status +func (r *ReconcileObservability) setNetworkObservabilityCondition(ctx context.Context, status operatorv1.ConditionStatus, reason, message string) error { + // Get the operator Network CR + network := &operatorv1.Network{} + if err := r.client.Get(ctx, types.NamespacedName{Name: NetworkCRName}, network); err != nil { + return fmt.Errorf("failed to get operator Network CR: %w", err) + } + + if condition := operatorv1helpers.FindOperatorCondition(network.Status.Conditions, NetworkObservabilityDeployed); condition != nil && + condition.Status == status && + condition.Reason == reason { + // Already set with same status and reason, no need to update + klog.V(4).Infof("Network Observability condition already set to %s with reason %s", status, reason) + return nil + } + + // Create the condition to add/update + operatorv1helpers.SetOperatorCondition(&network.Status.Conditions, operatorv1.OperatorCondition{ + Type: NetworkObservabilityDeployed, + Status: status, + Reason: reason, + Message: message, + }) + + // Update the status using controller-runtime client + if err := r.client.Status().Update(ctx, network); err != nil { + return fmt.Errorf("failed to update operator Network status: %w", err) + } + + klog.Infof("Set Network Observability condition to %s: %s", status, reason) + return nil +} + +// shouldInstallNetworkObservability returns true if Network Observability should be installed. +// Valid values of network.Spec.NetworkObservability.InstallationPolicy: "", "InstallAndEnable", "NoAction" +// "NoAction": skip installation (user opted out) +// "InstallAndEnable": install Network Observability once (even on SNO clusters), do not reinstall if already deployed +// "": install Network Observability once (opt-out model), except for SNO clusters, do not reinstall if already deployed +// SNO (Single Node OpenShift) clusters: skip installation by default unless explicitly set to "InstallAndEnable" +func (r *ReconcileObservability) shouldInstallNetworkObservability(ctx context.Context) (bool, error) { + // Get Network CR information + var network configv1.Network + if err := r.client.Get(ctx, types.NamespacedName{Name: NetworkCRName}, &network); err != nil { + if errors.IsNotFound(err) { + return false, nil + } + return false, err + } + + value := network.Spec.NetworkObservability.InstallationPolicy + + // Explicit disable + if value == configv1.NetworkObservabilityNoAction { + return false, nil + } + + // For both InstallAndEnable and default (empty): install once, do not reinstall + // Check if already deployed + deployed, err := r.wasNetworkObservabilityDeployed(ctx) + if err != nil { + return false, err + } + if deployed { + // Already deployed, do not reinstall + policyName := "default" + if value == configv1.NetworkObservabilityInstallAndEnable { + policyName = "InstallAndEnable" + } + klog.V(4).Infof("Network Observability already deployed (%s policy), skipping reinstallation", policyName) + return false, nil + } + + // Not yet deployed - determine if we should install based on policy and topology + // InstallAndEnable: install regardless of topology + if value == configv1.NetworkObservabilityInstallAndEnable { + return true, nil + } + + // Default behavior (empty string): install unless on SNO + isSNO, err := r.isSingleNodeCluster(ctx) + if err != nil { + return false, err + } + + if isSNO { + // SNO clusters: don't install by default + return false, nil + } + + // Non-SNO clusters: install by default (opt-out model) + return true, nil +} + +// isSingleNodeCluster returns true if the cluster is a Single Node OpenShift (SNO) cluster. +// A cluster is SNO if ControlPlaneTopology is SingleReplica. +func (r *ReconcileObservability) isSingleNodeCluster(ctx context.Context) (bool, error) { + infra := &configv1.Infrastructure{} + if err := r.client.Get(ctx, types.NamespacedName{Name: "cluster"}, infra); err != nil { + return false, err + } + + return infra.Status.ControlPlaneTopology == configv1.SingleReplicaTopologyMode, nil +} + +// isNetObservOperatorInstalled checks if the Network Observability Operator is installed +// by verifying both the FlowCollector CRD existence and the installation status via OLM. +// It checks both OLMv1 (ClusterExtension) and OLMv0 (ClusterServiceVersion) to determine +// installation status. +// Returns three values: +// - installed: true if the operator is fully installed +// - clusterExtensionExists: true if a ClusterExtension resource exists (relevant for OLMv1) +// - err: error if there was a problem checking the installation +func (r *ReconcileObservability) isNetObservOperatorInstalled(ctx context.Context) (installed bool, clusterExtensionExists bool, err error) { + // Check if the FlowCollector CRD exists + crd := &unstructured.Unstructured{} + crd.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "apiextensions.k8s.io", + Version: "v1", + Kind: "CustomResourceDefinition", + }) + + err = r.client.Get(ctx, types.NamespacedName{ + Name: "flowcollectors.flows.netobserv.io", + }, crd) + + crdExists := true + if err != nil { + if errors.IsNotFound(err) { + crdExists = false + } else { + return false, false, err + } + } + + // Check OLMv1 (ClusterExtension) installation status + olmv1Installed, olmv1CEExists, olmv1Err := r.checkOLMv1Installation(ctx) + if olmv1Err != nil { + // Installation error from OLMv1 + return false, olmv1CEExists, fmt.Errorf("OLMv1 installation error: %w", olmv1Err) + } + + // Check OLMv0 (ClusterServiceVersion/Subscription) installation status + olmv0Installed, olmv0Err := r.checkOLMv0Installation(ctx) + if olmv0Err != nil && !errors.IsNotFound(olmv0Err) { + // Installation error from OLMv0 + return false, olmv1CEExists, fmt.Errorf("OLMv0 installation error: %w", olmv0Err) + } + + // If CRD doesn't exist but either OLM installation is present, this is an error condition + if !crdExists { + if olmv0Installed || olmv1Installed { + olmVersion := "OLMv0" + if olmv1Installed { + olmVersion = "OLMv1" + } + return false, olmv1CEExists, fmt.Errorf("network Observability Operator was deployed via %s but FlowCollector CRD is missing (manually removed)", olmVersion) + } + // If CRD doesn't exist and no OLM installation, operator is not installed + return false, olmv1CEExists, nil + } + + if olmv1Installed { + klog.V(4).Info("Network Observability Operator installed via OLMv1 (ClusterExtension)") + return true, true, nil + } + + if olmv0Installed { + klog.V(4).Info("Network Observability Operator installed via OLMv0 (ClusterServiceVersion)") + return true, false, nil + } + + // CRD exists but neither OLMv0 nor OLMv1 shows a successful installation + return false, olmv1CEExists, fmt.Errorf("FlowCollector CRD is present but could not identify how Network Observability Operator was installed (neither OLMv1 ClusterExtension nor OLMv0 ClusterServiceVersion found)") +} + +// checkOLMv1Installation checks if the operator is installed via OLMv1 (ClusterExtension) +// Returns three values: +// - installed: true if the operator is fully installed via OLMv1 +// - clusterExtensionExists: true if the ClusterExtension resource exists (regardless of status) +// - err: error if there was a problem checking the installation +func (r *ReconcileObservability) checkOLMv1Installation(ctx context.Context) (installed bool, clusterExtensionExists bool, err error) { + clusterExtension := &unstructured.Unstructured{} + clusterExtension.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "olm.operatorframework.io", + Version: "v1", + Kind: "ClusterExtension", + }) + + if err := r.client.Get(ctx, types.NamespacedName{Name: "netobserv-operator"}, clusterExtension); err != nil { + if errors.IsNotFound(err) { + return false, false, nil + } + return false, false, err + } + + // ClusterExtension exists + // Check its status conditions + conditions, found, err := unstructured.NestedSlice(clusterExtension.Object, "status", "conditions") + if err != nil { + return false, true, fmt.Errorf("failed to get ClusterExtension status conditions: %w", err) + } + if !found { + return false, true, fmt.Errorf("ClusterExtension exists but has no status conditions") + } + + // Check for "Installed" condition + for _, cond := range conditions { + condMap, ok := cond.(map[string]interface{}) + if !ok { + continue + } + condType, _, _ := unstructured.NestedString(condMap, "type") + condStatus, _, _ := unstructured.NestedString(condMap, "status") + condReason, _, _ := unstructured.NestedString(condMap, "reason") + condMessage, _, _ := unstructured.NestedString(condMap, "message") + + if condType == "Installed" { + switch condStatus { + case "True": + return true, true, nil + case "False": + return false, true, fmt.Errorf("ClusterExtension installation failed: %s - %s", condReason, condMessage) + default: + // Status is "Unknown" or other - not yet installed + return false, true, nil + } + } + } + + // ClusterExtension exists but no "Installed" condition found + return false, true, nil +} + +// checkOLMv0Installation checks if the operator is installed via OLMv0 (CSV) +func (r *ReconcileObservability) checkOLMv0Installation(ctx context.Context) (bool, error) { + csvList := &unstructured.UnstructuredList{} + csvList.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "operators.coreos.com", + Version: "v1alpha1", + Kind: "ClusterServiceVersion", + }) + + if err := r.client.List(ctx, csvList, crclient.InNamespace(OperatorNamespace)); err != nil { + if errors.IsNotFound(err) { + return false, nil + } + return false, err + } + + // Look for netobserv operator CSV + for _, item := range csvList.Items { + name := item.GetName() + // CSV names typically follow pattern: netobserv-operator.v1.2.3 + if strings.HasPrefix(name, "netobserv-operator") { + // Check the CSV phase + phase, found, err := unstructured.NestedString(item.Object, "status", "phase") + if err != nil { + return false, fmt.Errorf("failed to get ClusterServiceVersion status phase: %w", err) + } + if !found { + return false, fmt.Errorf("ClusterServiceVersion exists but has no status phase") + } + + switch phase { + case "Succeeded": + return true, nil + case "Failed": + reason, _, _ := unstructured.NestedString(item.Object, "status", "reason") + message, _, _ := unstructured.NestedString(item.Object, "status", "message") + return false, fmt.Errorf("ClusterServiceVersion installation failed: %s - %s", reason, message) + default: + // Other phases (Installing, Pending, Replacing, Deleting, etc.) - not yet installed + return false, nil + } + } + } + + // No CSV found + return false, nil +} + +// applyManifest reads a YAML file and applies all resources using server-side apply +func (r *ReconcileObservability) applyManifest(ctx context.Context, yamlPath, description string) error { + yamlBytes, err := os.ReadFile(yamlPath) + if err != nil { + return fmt.Errorf("failed to read %s manifest %s: %w", description, yamlPath, err) + } + + dec := yaml.NewYAMLOrJSONDecoder(bytes.NewReader(yamlBytes), 4096) + for { + obj := &unstructured.Unstructured{} + if err := dec.Decode(obj); err != nil { + if err == io.EOF { + break + } + return err + } + if obj.GetKind() == "" { + continue + } + obj.SetManagedFields(nil) + + // Marshal object to JSON for RawPatch + data, err := obj.MarshalJSON() + if err != nil { + return fmt.Errorf("failed to marshal %s %s: %w", obj.GetKind(), obj.GetName(), err) + } + + // Use RawPatch with ApplyPatchType to avoid deprecated crclient.Apply + patch := crclient.RawPatch(types.ApplyPatchType, data) + if err := r.client.Patch(ctx, obj, patch, &crclient.PatchOptions{ + FieldManager: "cno-observability-controller", + }); err != nil { + return fmt.Errorf("failed to apply %s %s: %w", obj.GetKind(), obj.GetName(), err) + } + klog.Infof("Applied %s %s", description, obj.GetName()) + } + klog.Infof("Successfully applied %s", description) + return nil +} + +func (r *ReconcileObservability) installNetObservOperator(ctx context.Context) error { + return r.applyManifest(ctx, OperatorYAML, "Network Observability Operator") +} + +// isFlowCollectorExists returns true if a FlowCollector instance exists. +// Note: FlowCollector is a cluster-scoped singleton resource and can only be named "cluster". +func (r *ReconcileObservability) isFlowCollectorExists(ctx context.Context) (bool, error) { + flowCollector := &unstructured.Unstructured{} + flowCollector.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "flows.netobserv.io", + Version: FlowCollectorVersion, + Kind: "FlowCollector", + }) + + err := r.client.Get(ctx, types.NamespacedName{Name: FlowCollectorName}, flowCollector) + if err != nil { + if errors.IsNotFound(err) { + return false, nil + } + return false, err + } + + return true, nil +} + +func (r *ReconcileObservability) createFlowCollector(ctx context.Context) error { + // Ensure the netobserv namespace exists before applying manifests. + ns := &corev1.Namespace{} + if err := r.client.Get(ctx, types.NamespacedName{Name: NetObservNamespace}, ns); err != nil { + if errors.IsNotFound(err) { + if err := r.client.Create(ctx, &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: NetObservNamespace}, + }); err != nil { + return fmt.Errorf("failed to create namespace %s: %w", NetObservNamespace, err) + } + klog.Infof("Created namespace %s", NetObservNamespace) + } else { + return err + } + } + + return r.applyManifest(ctx, FlowCollectorYAML, "FlowCollector") +} diff --git a/pkg/controller/observability/observability_controller_test.go b/pkg/controller/observability/observability_controller_test.go new file mode 100644 index 0000000000..1b18f07c3b --- /dev/null +++ b/pkg/controller/observability/observability_controller_test.go @@ -0,0 +1,1780 @@ +package observability + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + . "github.com/onsi/gomega" + + configv1 "github.com/openshift/api/config/v1" + operatorv1 "github.com/openshift/api/operator/v1" + "github.com/openshift/library-go/pkg/operator/configobserver/featuregates" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +// Helper functions for creating test resources + +func createTestNetwork(name string, value string) *configv1.Network { + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + }, + } + + if value != "" { + network.Spec.NetworkObservability = configv1.NetworkObservabilitySpec{ + InstallationPolicy: configv1.NetworkObservabilityInstallationPolicy(value), + } + } + + return network +} + +func createTestOperatorNetwork(name string) *operatorv1.Network { + return &operatorv1.Network{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + }, + } +} + +func createTestOperatorNetworkWithDeployedCondition(name string) *operatorv1.Network { + return &operatorv1.Network{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + }, + Status: operatorv1.NetworkStatus{ + OperatorStatus: operatorv1.OperatorStatus{ + Conditions: []operatorv1.OperatorCondition{ + { + Type: NetworkObservabilityDeployed, + Status: operatorv1.ConditionTrue, + Reason: "DeploymentComplete", + Message: "Network Observability has been deployed", + LastTransitionTime: metav1.Now(), + }, + }, + }, + }, + } +} + +func createTestFlowCollector(name string) *unstructured.Unstructured { + fc := &unstructured.Unstructured{} + fc.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "flows.netobserv.io", + Version: FlowCollectorVersion, + Kind: "FlowCollector", + }) + fc.SetName(name) + return fc +} + +func createTestNamespace(name string) *corev1.Namespace { + return &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + }, + } +} + +func createTestInfrastructure(topology configv1.TopologyMode) *configv1.Infrastructure { + return &configv1.Infrastructure{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cluster", + }, + Status: configv1.InfrastructureStatus{ + ControlPlaneTopology: topology, + }, + } +} + +func createTestCRD(name string) *unstructured.Unstructured { + crd := &unstructured.Unstructured{} + crd.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "apiextensions.k8s.io", + Version: "v1", + Kind: "CustomResourceDefinition", + }) + crd.SetName(name) + return crd +} + +func createTestClusterExtension(t *testing.T, name string, installed bool) *unstructured.Unstructured { + t.Helper() + ce := &unstructured.Unstructured{} + ce.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "olm.operatorframework.io", + Version: "v1", + Kind: "ClusterExtension", + }) + ce.SetName(name) + + // Set status conditions + conditions := []interface{}{ + map[string]interface{}{ + "type": "Installed", + "status": func() string { + if installed { + return "True" + } + return "False" + }(), + "reason": "InstallSucceeded", + "message": "ClusterExtension installed successfully", + }, + } + if err := unstructured.SetNestedSlice(ce.Object, conditions, "status", "conditions"); err != nil { + t.Fatalf("Failed to set status conditions: %v", err) + } + return ce +} + +func createTestCSV(name string, succeeded bool) *unstructured.Unstructured { + csv := &unstructured.Unstructured{} + csv.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "operators.coreos.com", + Version: "v1alpha1", + Kind: "ClusterServiceVersion", + }) + csv.SetName(name + ".v1.0.0") + csv.SetNamespace(OperatorNamespace) + + // Set status phase + phase := "Succeeded" + if !succeeded { + phase = "Failed" + } + _ = unstructured.SetNestedField(csv.Object, phase, "status", "phase") + + return csv +} + +func createTempManifest(t *testing.T, content string) string { + t.Helper() + tmpDir := t.TempDir() + filePath := filepath.Join(tmpDir, "manifest.yaml") + err := os.WriteFile(filePath, []byte(content), 0644) + if err != nil { + t.Fatalf("Failed to create temp manifest: %v", err) + } + return filePath +} + +// Helper function to create a feature gate with NetworkObservabilityInstall enabled +func createEnabledFeatureGate() featuregates.FeatureGate { + return featuregates.NewFeatureGate( + []configv1.FeatureGateName{"NetworkObservabilityInstall"}, + []configv1.FeatureGateName{}, + ) +} + +// Test shouldInstallNetworkObservability() + +func TestShouldInstallNetworkObservability_NilNonSNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + // NetworkObservability not set: Default behavior should install on non-SNO + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeTrue()) +} + +func TestShouldInstallNetworkObservability_NilSNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + // NetworkObservability not set: Default behavior should NOT install on SNO + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.SingleReplicaTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeFalse()) +} + +func TestShouldInstallNetworkObservability_ExplicitInstallAndEnableNonSNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + NetworkObservability: configv1.NetworkObservabilitySpec{ + InstallationPolicy: configv1.NetworkObservabilityInstallAndEnable, + }, + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeTrue()) +} + +func TestShouldInstallNetworkObservability_ExplicitInstallAndEnableSNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + NetworkObservability: configv1.NetworkObservabilitySpec{ + InstallationPolicy: configv1.NetworkObservabilityInstallAndEnable, // Explicit InstallAndEnable: install even on SNO + }, + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.SingleReplicaTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeTrue()) +} + +func TestShouldInstallNetworkObservability_ExplicitNoAction(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + NetworkObservability: configv1.NetworkObservabilitySpec{ + InstallationPolicy: configv1.NetworkObservabilityNoAction, + }, + }, + } + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeFalse()) +} + +func TestShouldInstallNetworkObservability_EmptyStringNonSNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + NetworkObservability: configv1.NetworkObservabilitySpec{ + InstallationPolicy: "", // Empty string: default behavior (install on non-SNO) + }, + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeTrue()) +} + +// Test isSingleNodeCluster() + +func TestIsSingleNodeCluster_SNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + infra := createTestInfrastructure(configv1.SingleReplicaTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(infra).Build() + r := &ReconcileObservability{client: client} + + isSNO, err := r.isSingleNodeCluster(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(isSNO).To(BeTrue()) +} + +func TestIsSingleNodeCluster_HighlyAvailable(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(infra).Build() + r := &ReconcileObservability{client: client} + + isSNO, err := r.isSingleNodeCluster(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(isSNO).To(BeFalse()) +} + +// Test Reconcile() - Main Controller Logic + +func TestReconcile_IgnoresNonClusterNetwork(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + network := createTestNetwork("not-cluster", "InstallAndEnable") + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network).Build() + + r := &ReconcileObservability{ + client: client, + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "not-cluster"}} + result, err := r.Reconcile(context.TODO(), req) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +func TestReconcile_SkipsWhenDisabled(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + // Explicitly set to false (opt-out) + network := createTestNetwork("cluster", "NoAction") + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network).Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + result, err := r.Reconcile(context.TODO(), req) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +func TestReconcile_InstallsWhenNil(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Create network with no NetworkObservability field (defaults to enabled on non-SNO) + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cluster", + }, + Spec: configv1.NetworkSpec{ + // NetworkObservability not set: defaults to enabled on non-SNO + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + operatorNs := createTestNamespace(OperatorNamespace) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra, operatorNs).Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + result, err := r.Reconcile(context.TODO(), req) + + // When nil, controller should try to install (opt-out behavior) + // This will fail because the manifest doesn't exist, but it requeues instead of erroring + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(result.RequeueAfter).To(Equal(requeueAfterOLM)) +} + +func TestReconcile_SkipsInstallWhenNilOnSNO(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Create network with no NetworkObservability field on SNO cluster + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cluster", + }, + Spec: configv1.NetworkSpec{ + // NetworkObservability not set: defaults to disabled on SNO + }, + } + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.SingleReplicaTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + result, err := r.Reconcile(context.TODO(), req) + + // On SNO with nil, controller should skip installation + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) + + // Verify that the operator namespace was NOT created + ns := &corev1.Namespace{} + nsErr := client.Get(context.TODO(), types.NamespacedName{Name: OperatorNamespace}, ns) + g.Expect(nsErr).To(HaveOccurred()) + g.Expect(nsErr.Error()).To(ContainSubstring("not found")) +} + +func TestReconcile_IgnoresNotFound(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{ + client: client, + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + result, err := r.Reconcile(context.TODO(), req) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// Test isNetObservOperatorInstalled() + +func TestIsNetObservOperatorInstalled_True(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd, clusterExtension).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(installed).To(BeTrue()) + g.Expect(ceExists).To(BeTrue()) +} + +func TestIsNetObservOperatorInstalled_False(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(installed).To(BeFalse()) + g.Expect(ceExists).To(BeFalse()) +} + +func TestIsNetObservOperatorInstalled_Multiple(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + // Create multiple CRDs, but only the FlowCollector one should matter + crd1 := createTestCRD("other-crds.example.com") + crd2 := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd1, crd2, clusterExtension).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(installed).To(BeTrue()) + g.Expect(ceExists).To(BeTrue()) +} + +func TestIsNetObservOperatorInstalled_CRDExistsButNoOLM(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + // Create only CRD, no ClusterExtension or CSV + crd := createTestCRD("flowcollectors.flows.netobserv.io") + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + // Should return error because CRD exists but no OLM installation found + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("FlowCollector CRD is present but could not identify")) + g.Expect(installed).To(BeFalse()) + g.Expect(ceExists).To(BeFalse()) +} + +func TestIsNetObservOperatorInstalled_OLMv1InstallationFailed(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + crd := createTestCRD("flowcollectors.flows.netobserv.io") + + // Create ClusterExtension with Installed=False (installation failed) + ce := &unstructured.Unstructured{} + ce.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "olm.operatorframework.io", + Version: "v1", + Kind: "ClusterExtension", + }) + ce.SetName("netobserv-operator") + conditions := []interface{}{ + map[string]interface{}{ + "type": "Installed", + "status": "False", + "reason": "InstallationFailed", + "message": "Failed to install operator bundle", + }, + } + if err := unstructured.SetNestedSlice(ce.Object, conditions, "status", "conditions"); err != nil { + t.Fatalf("Failed to set status conditions: %v", err) + } + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd, ce).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + // Should return error because installation failed + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("OLMv1 installation error")) + g.Expect(err.Error()).To(ContainSubstring("ClusterExtension installation failed")) + g.Expect(installed).To(BeFalse()) + g.Expect(ceExists).To(BeTrue()) +} + +func TestIsNetObservOperatorInstalled_OLMv1NotInstalledYet(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + crd := createTestCRD("flowcollectors.flows.netobserv.io") + + // Create ClusterExtension with Installed=Unknown (not installed yet) + ce := &unstructured.Unstructured{} + ce.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "olm.operatorframework.io", + Version: "v1", + Kind: "ClusterExtension", + }) + ce.SetName("netobserv-operator") + conditions := []interface{}{ + map[string]interface{}{ + "type": "Installed", + "status": "Unknown", + "reason": "Installing", + "message": "Installing operator bundle", + }, + } + if err := unstructured.SetNestedSlice(ce.Object, conditions, "status", "conditions"); err != nil { + t.Fatalf("Failed to set status conditions: %v", err) + } + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd, ce).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + // Should return an error because the controller could not identify how NOO was installed + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("could not identify how Network Observability Operator was installed")) + g.Expect(installed).To(BeFalse()) + g.Expect(ceExists).To(BeTrue()) +} + +func TestIsNetObservOperatorInstalled_CRDMissingButOLMv1Present(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + // No CRD, but ClusterExtension exists + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(clusterExtension).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + // Should return error because operator was deployed but CRD is missing + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("network Observability Operator was deployed via OLMv1 but FlowCollector CRD is missing")) + g.Expect(installed).To(BeFalse()) + g.Expect(ceExists).To(BeTrue()) +} + +func TestIsNetObservOperatorInstalled_CRDMissingButOLMv0Present(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + // No CRD, but CSV exists + csv := createTestCSV("netobserv-operator", true) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(csv).Build() + + r := &ReconcileObservability{client: client} + + installed, ceExists, err := r.isNetObservOperatorInstalled(context.TODO()) + + // Should return error because operator was deployed but CRD is missing + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("network Observability Operator was deployed via OLMv0 but FlowCollector CRD is missing")) + g.Expect(installed).To(BeFalse()) + g.Expect(ceExists).To(BeFalse()) +} + +// Test isFlowCollectorExists() + +func TestIsFlowCollectorExists_True(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + flowCollector := createTestFlowCollector(FlowCollectorName) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(flowCollector).Build() + + r := &ReconcileObservability{client: client} + + exists, err := r.isFlowCollectorExists(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(exists).To(BeTrue()) +} + +func TestIsFlowCollectorExists_False(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + exists, err := r.isFlowCollectorExists(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(exists).To(BeFalse()) +} + +func TestIsFlowCollectorExists_OnlyChecksCluster(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + // Create a FlowCollector with a different name + fcOther := createTestFlowCollector("other") + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(fcOther).Build() + + r := &ReconcileObservability{client: client} + + exists, err := r.isFlowCollectorExists(context.TODO()) + + // Should return false because we only check for "cluster" + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(exists).To(BeFalse()) +} + +// Test createFlowCollector() - Note: Full testing requires real manifest files + +func TestCreateFlowCollector_ManifestNotFound(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + // Test with non-existent manifest by calling applyManifest directly + err := r.applyManifest(context.TODO(), "/non/existent/path.yaml", "test") + + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("failed to read")) +} + +// Test installNetObservOperator() + +func TestInstallNetObservOperator_ManifestNotFound(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + // Test applyManifest with non-existent path directly + err := r.applyManifest(context.TODO(), "/non/existent/operator.yaml", "Network Observability Operator") + + g.Expect(err).To(HaveOccurred()) + g.Expect(err.Error()).To(ContainSubstring("failed to read")) +} + +// Test applyManifest() + +func TestApplyManifest_SingleResource(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + manifestContent := ` +apiVersion: v1 +kind: Namespace +metadata: + name: test-namespace +` + manifestPath := createTempManifest(t, manifestContent) + + err := r.applyManifest(context.TODO(), manifestPath, "test resource") + + g.Expect(err).NotTo(HaveOccurred()) +} + +func TestApplyManifest_MultipleResources(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + manifestContent := ` +apiVersion: v1 +kind: Namespace +metadata: + name: test-namespace-1 +--- +apiVersion: v1 +kind: Namespace +metadata: + name: test-namespace-2 +` + manifestPath := createTempManifest(t, manifestContent) + + err := r.applyManifest(context.TODO(), manifestPath, "test resources") + + g.Expect(err).NotTo(HaveOccurred()) +} + +func TestApplyManifest_EmptyDocuments(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + manifestContent := `--- +--- +` + manifestPath := createTempManifest(t, manifestContent) + + err := r.applyManifest(context.TODO(), manifestPath, "empty resources") + + // Should not error on empty documents + g.Expect(err).NotTo(HaveOccurred()) +} + +func TestApplyManifest_InvalidYAML(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + + client := fake.NewClientBuilder().WithScheme(scheme).Build() + + r := &ReconcileObservability{client: client} + + manifestContent := ` +invalid: yaml: content: + - broken + indentation +` + manifestPath := createTempManifest(t, manifestContent) + + err := r.applyManifest(context.TODO(), manifestPath, "invalid resource") + + g.Expect(err).To(HaveOccurred()) +} + +// Integration Tests + +// TestReconcile_SkipsFlowCollectorWhenExists tests that reconciliation +// doesn't try to create FlowCollector if it already exists +func TestReconcile_SkipsFlowCollectorWhenExists(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + operatorNetwork := createTestOperatorNetwork("cluster") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + flowCollector := createTestFlowCollector(FlowCollectorName) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, crd, clusterExtension, flowCollector). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + result, err := r.Reconcile(context.TODO(), req) + + // Should complete without error + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// TestReconcile_SkipsInstallWhenExists tests that reconciliation +// doesn't try to install operator if it already exists +func TestReconcile_SkipsInstallWhenExists(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + operatorNs := createTestNamespace(OperatorNamespace) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, crd, clusterExtension, operatorNs). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Since operator is already installed, it should proceed to FlowCollector creation + // which will fail (manifest doesn't exist) but will requeue instead of erroring + result, err := r.Reconcile(context.TODO(), req) + + // We expect no error, just a requeue + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(result.RequeueAfter).To(Equal(requeueAfterStandard)) +} + +// Edge Case Tests + +// TestReconcile_MultipleInvocations tests that multiple reconciliations +// handle idempotency correctly +func TestReconcile_MultipleInvocations(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + operatorNetwork := createTestOperatorNetwork("cluster") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + flowCollector := createTestFlowCollector(FlowCollectorName) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, crd, clusterExtension, flowCollector). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // First reconciliation + result1, err1 := r.Reconcile(context.TODO(), req) + g.Expect(err1).NotTo(HaveOccurred()) + g.Expect(result1).To(Equal(reconcile.Result{})) + + // Second reconciliation should be idempotent + result2, err2 := r.Reconcile(context.TODO(), req) + g.Expect(err2).NotTo(HaveOccurred()) + g.Expect(result2).To(Equal(reconcile.Result{})) + + // Results should be the same + g.Expect(result1).To(Equal(result2)) +} + +// TestReconcile_OperatorNotReady tests reconciliation when operator exists +// but is not ready yet +func TestReconcile_OperatorNotReady(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + // CSV exists but not in Succeeded phase + clusterExtension := createTestClusterExtension(t, "netobserv-operator", false) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, crd, clusterExtension). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + result, err := r.Reconcile(ctx, req) + + // Controller returns no error, but should requeue after failing FlowCollector creation + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result.RequeueAfter).To(Equal(requeueAfterStandard)) +} + +// TestReconcile_FlowCollectorDeleted tests that reconciliation does not recreate +// FlowCollector if it was previously deployed and then deleted (default policy) +func TestReconcile_FlowCollectorDeleted(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Create network with the deployed condition set (simulating previous successful deployment) + // FlowCollector is NOT present (deleted) + // Use default policy (empty string) - which should NOT reinstall after deployment + network := createTestNetwork("cluster", "") + operatorNetwork := createTestOperatorNetworkWithDeployedCondition("cluster") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, crd, clusterExtension). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should skip everything since deployment condition is set + result, err := r.Reconcile(context.TODO(), req) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// TestReconcile_OperatorDeleted tests that operator is not reinstalled after deletion if previously deployed (default policy) +func TestReconcile_OperatorDeleted(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Create network with the deployed condition set (simulating previous successful deployment) + // Operator subscription and CSV are NOT present (simulating deletion) + // Use default policy (empty string) - which should NOT reinstall after deployment + network := createTestNetwork("cluster", "") + operatorNetwork := createTestOperatorNetworkWithDeployedCondition("cluster") + flowCollector := createTestFlowCollector(FlowCollectorName) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, flowCollector). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should skip everything since deployment condition is set + result, err := r.Reconcile(context.TODO(), req) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// TestReconcile_BothDeleted tests that nothing is reinstalled when both operator and FlowCollector are deleted (default policy) +func TestReconcile_BothDeleted(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Create network with the deployed condition set (simulating previous successful deployment) + // Neither operator nor FlowCollector are present (both deleted) + // Use default policy (empty string) - which should NOT reinstall after deployment + network := createTestNetwork("cluster", "") + operatorNetwork := createTestOperatorNetworkWithDeployedCondition("cluster") + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should skip everything since deployment condition is set + result, err := r.Reconcile(context.TODO(), req) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// TestReconcile_InstallAndEnable_DoesNotReinstallWhenDeployed tests that InstallAndEnable +// policy does NOT reinstall when already deployed (same behavior as default policy) +func TestReconcile_InstallAndEnable_DoesNotReinstallWhenDeployed(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Create network with InstallAndEnable policy and deployed condition set + // FlowCollector is NOT present (deleted) + // InstallAndEnable should NOT reinstall after deployment + network := createTestNetwork("cluster", string(configv1.NetworkObservabilityInstallAndEnable)) + operatorNetwork := createTestOperatorNetworkWithDeployedCondition("cluster") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, crd, clusterExtension). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should skip everything since deployment condition is set + result, err := r.Reconcile(context.TODO(), req) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// TestShouldInstallNetworkObservability_InstallAndEnable_AlreadyDeployed tests that +// InstallAndEnable returns false when already deployed +func TestShouldInstallNetworkObservability_InstallAndEnable_AlreadyDeployed(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + NetworkObservability: configv1.NetworkObservabilitySpec{ + InstallationPolicy: configv1.NetworkObservabilityInstallAndEnable, + }, + }, + } + operatorNetwork := createTestOperatorNetworkWithDeployedCondition("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(network, operatorNetwork, infra).Build() + r := &ReconcileObservability{client: client} + + result, err := r.shouldInstallNetworkObservability(context.TODO()) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(BeFalse()) // Should NOT install when already deployed +} + +// TestReconcile_NetworkCRUpdated tests that reconciliation handles +// Network CR updates correctly +func TestReconcile_NetworkCRUpdated(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + // Start with disabled + network := createTestNetwork("cluster", "NoAction") + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, infra). + WithStatusSubresource(&configv1.Network{}, &operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // First reconciliation - should skip + result1, err1 := r.Reconcile(context.TODO(), req) + g.Expect(err1).NotTo(HaveOccurred()) + g.Expect(result1).To(Equal(reconcile.Result{})) + + // Update Network CR to enable observability + network.Spec.NetworkObservability = configv1.NetworkObservabilitySpec{ + InstallationPolicy: configv1.NetworkObservabilityInstallAndEnable, + } + err := client.Update(context.TODO(), network) + g.Expect(err).NotTo(HaveOccurred()) + + // Second reconciliation - should now try to install + // This will fail because manifest doesn't exist, but will requeue instead of erroring + result2, err2 := r.Reconcile(context.TODO(), req) + + // Should requeue, not error + g.Expect(err2).ToNot(HaveOccurred()) + g.Expect(result2.RequeueAfter).To(Equal(requeueAfterOLM)) +} + +// Error Scenario Tests + +// TestReconcile_PartialFailure_OperatorInstallFails tests recovery +// when operator installation fails +func TestReconcile_PartialFailure_OperatorInstallFails(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, infra). + WithStatusSubresource(&configv1.Network{}, &operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should requeue when install fails (manifest doesn't exist) + result, err := r.Reconcile(context.TODO(), req) + + // Should requeue, not error + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(result.RequeueAfter).To(Equal(requeueAfterOLM)) +} + +// TestReconcile_RecoveryAfterOperatorBecomesReady tests that reconciliation +// continues after operator becomes ready +func TestReconcile_RecoveryAfterOperatorBecomesReady(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + // Start with CSV in Installing phase + clusterExtension := createTestClusterExtension(t, "netobserv-operator", false) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, crd, clusterExtension). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // First reconciliation will fail creating FlowCollector (returns no error but RequeueAfter=30s) + ctx1, cancel1 := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel1() + + result1, err1 := r.Reconcile(ctx1, req) + g.Expect(err1).NotTo(HaveOccurred()) + g.Expect(result1.RequeueAfter).To(Equal(requeueAfterStandard)) + + // Update ClusterExtension to Installed status + conditions := []interface{}{ + map[string]interface{}{ + "type": "Installed", + "status": "True", + "reason": "InstallSucceeded", + "message": "ClusterExtension installed successfully", + }, + } + if err := unstructured.SetNestedSlice(clusterExtension.Object, conditions, "status", "conditions"); err != nil { + t.Fatalf("Failed to set status conditions: %v", err) + } + err := client.Update(context.TODO(), clusterExtension) + g.Expect(err).NotTo(HaveOccurred()) + + // Second reconciliation should proceed past operator wait + // and attempt to create FlowCollector (which will fail due to missing manifest) + ctx2, cancel2 := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel2() + + result2, err2 := r.Reconcile(ctx2, req) + + // Should requeue after failing to read FlowCollector manifest + g.Expect(err2).ToNot(HaveOccurred()) + g.Expect(result2.RequeueAfter).To(Equal(requeueAfterStandard)) +} + +// Performance/Stress Tests + +// TestReconcile_ConcurrentReconciliations tests that multiple concurrent +// reconciliations don't cause issues +func TestReconcile_ConcurrentReconciliations(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + flowCollector := createTestFlowCollector(FlowCollectorName) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, crd, clusterExtension, flowCollector). + WithStatusSubresource(&configv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Run 5 concurrent reconciliations + errChan := make(chan error, 5) + for i := 0; i < 5; i++ { + go func() { + _, err := r.Reconcile(context.TODO(), req) + errChan <- err + }() + } + + // Wait for all to complete and collect errors + var unexpectedErrors []error + for i := 0; i < 5; i++ { + if err := <-errChan; err != nil { + // Filter out 409 conflict errors which are expected when multiple + // goroutines try to update the same resource status concurrently + if !errors.IsConflict(err) { + unexpectedErrors = append(unexpectedErrors, err) + } + } + } + + // Assert no unexpected errors occurred (safe to do in main test goroutine) + g.Expect(unexpectedErrors).To(BeEmpty(), "All concurrent reconciliations should complete without unexpected errors") +} + +// Status Manager Tests + +// TestReconcile_SetsConditionFalseOnError tests that NetworkObservabilityDeployed condition is set to False on errors +func TestReconcile_SetsConditionFalseOnError(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, infra). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should fail trying to install operator (manifest doesn't exist) + result, err := r.Reconcile(context.TODO(), req) + + // Should requeue without error + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(result.RequeueAfter).To(Equal(requeueAfterOLM)) + + // Verify that NetworkObservabilityDeployed condition is set to False + updatedNetwork := &operatorv1.Network{} + err = client.Get(context.TODO(), types.NamespacedName{Name: "cluster"}, updatedNetwork) + g.Expect(err).NotTo(HaveOccurred()) + + conditionFound := false + for _, condition := range updatedNetwork.Status.Conditions { + if condition.Type == NetworkObservabilityDeployed { + conditionFound = true + g.Expect(condition.Status).To(Equal(operatorv1.ConditionFalse)) + g.Expect(condition.Reason).To(Equal("DeploymentFailed")) + g.Expect(condition.Message).To(ContainSubstring("Failed to install Network Observability Operator")) + break + } + } + g.Expect(conditionFound).To(BeTrue(), "NetworkObservabilityDeployed condition should be set") +} + +// TestReconcile_SetsConditionTrueOnSuccess tests that NetworkObservabilityDeployed condition is set to True on success +func TestReconcile_SetsConditionTrueOnSuccess(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "InstallAndEnable") + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + crd := createTestCRD("flowcollectors.flows.netobserv.io") + clusterExtension := createTestClusterExtension(t, "netobserv-operator", true) + flowCollector := createTestFlowCollector(FlowCollectorName) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, infra, crd, clusterExtension, flowCollector). + WithStatusSubresource(&operatorv1.Network{}). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should succeed (FlowCollector already exists) + _, err := r.Reconcile(context.TODO(), req) + + g.Expect(err).NotTo(HaveOccurred()) + + // Verify that NetworkObservabilityDeployed condition is set to True + updatedNetwork := &operatorv1.Network{} + err = client.Get(context.TODO(), types.NamespacedName{Name: "cluster"}, updatedNetwork) + g.Expect(err).NotTo(HaveOccurred()) + + conditionFound := false + for _, condition := range updatedNetwork.Status.Conditions { + if condition.Type == NetworkObservabilityDeployed { + conditionFound = true + g.Expect(condition.Status).To(Equal(operatorv1.ConditionTrue)) + g.Expect(condition.Reason).To(Equal("DeploymentComplete")) + g.Expect(condition.Message).To(Equal("Network Observability has been deployed")) + break + } + } + g.Expect(conditionFound).To(BeTrue(), "NetworkObservabilityDeployed condition should be set") +} + +// TestReconcile_SkipsWhenNoActionPolicy tests that reconciliation skips when NoAction policy is set +func TestReconcile_SkipsWhenNoActionPolicy(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add corev1 to scheme: %v", err) + } + if err := operatorv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add operatorv1 to scheme: %v", err) + } + + network := createTestNetwork("cluster", "NoAction") // disabled + operatorNetwork := createTestOperatorNetwork("cluster") + infra := createTestInfrastructure(configv1.HighlyAvailableTopologyMode) + + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network, operatorNetwork, infra). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should succeed and skip installation + result, err := r.Reconcile(context.TODO(), req) + + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(reconcile.Result{})) +} + +// TestReconcile_RequeuesOnInfrastructureError tests that reconciliation requeues when Infrastructure lookup fails +func TestReconcile_RequeuesOnInfrastructureError(t *testing.T) { + g := NewGomegaWithT(t) + + scheme := runtime.NewScheme() + if err := configv1.AddToScheme(scheme); err != nil { + t.Fatalf("Failed to add configv1 to scheme: %v", err) + } + + // Create network with no NetworkObservability field (will trigger SNO check which needs Infrastructure) + network := &configv1.Network{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster"}, + Spec: configv1.NetworkSpec{ + // NetworkObservability not set: will trigger SNO check + }, + } + + // Don't add Infrastructure object - this will cause Get to fail + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(network). + Build() + + r := &ReconcileObservability{ + client: client, + featureGate: createEnabledFeatureGate(), + } + + req := reconcile.Request{NamespacedName: types.NamespacedName{Name: "cluster"}} + + // Reconciliation should requeue when checking Infrastructure fails + result, err := r.Reconcile(context.TODO(), req) + + // Should requeue without error (errors are logged and requeued) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(result.RequeueAfter).To(Equal(requeueAfterStandard)) +} + +func TestIsFeatureGateEnabled_NilFeatureGate(t *testing.T) { + g := NewGomegaWithT(t) + + r := &ReconcileObservability{featureGate: nil} + + // Should default to disabled when featureGate is nil + result := r.isFeatureGateEnabled() + g.Expect(result).To(BeFalse()) +} + +func TestIsFeatureGateEnabled_FeatureGateEnabled(t *testing.T) { + g := NewGomegaWithT(t) + + // Create a feature gate with NetworkObservabilityInstall enabled + fg := featuregates.NewFeatureGate( + []configv1.FeatureGateName{"NetworkObservabilityInstall"}, + []configv1.FeatureGateName{}, + ) + + r := &ReconcileObservability{featureGate: fg} + + result := r.isFeatureGateEnabled() + g.Expect(result).To(BeTrue()) +} + +func TestIsFeatureGateEnabled_FeatureGateDisabled(t *testing.T) { + g := NewGomegaWithT(t) + + // Create a feature gate with NetworkObservabilityInstall disabled + fg := featuregates.NewFeatureGate( + []configv1.FeatureGateName{}, + []configv1.FeatureGateName{"NetworkObservabilityInstall"}, + ) + + r := &ReconcileObservability{featureGate: fg} + + result := r.isFeatureGateEnabled() + g.Expect(result).To(BeFalse()) +} diff --git a/sample-cluster-network-config.yaml b/sample-cluster-network-config.yaml index 3b7a95e469..ab42447dce 100644 --- a/sample-cluster-network-config.yaml +++ b/sample-cluster-network-config.yaml @@ -5,7 +5,14 @@ metadata: spec: serviceNetwork: - "172.30.0.0/16" - clusterNetwork: + clusterNetwork: - cidr: "10.128.0.0/14" - hostPrefix: 23 + hostPrefix: 23 networkType: OVNKubernetes + networkObservability: + # installationPolicy controls Network Observability installation during cluster deployment (day-0). + # Valid values: "", "InstallAndEnable", "NoAction" + # Default (empty or omitted): enabled on multi-node clusters, disabled on SNO + # "InstallAndEnable": explicitly enable (even on SNO) + # "NoAction": explicitly disable + installationPolicy: InstallAndEnable