address feedback from review

Author: labrenbe

CHANGELOG.md

@@ -7,7 +7,7 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Add rolling upgrade support for upgrades between NiFi 2 versions ([#771]).
-- Added listener support for Nifi ([#784]).
+- Added Listener support for NiFi ([#784]).
 - Adds new telemetry CLI arguments and environment variables ([#782]).
   - Use `--file-log-max-files` (or `FILE_LOG_MAX_FILES`) to limit the number of log files kept.
   - Use `--file-log-rotation-period` (or `FILE_LOG_ROTATION_PERIOD`) to configure the frequency of rotation.

rust/operator-binary/src/controller.rs

@@ -140,9 +140,6 @@ pub enum Error {
     #[snafu(display("object defines no name"))]
     ObjectHasNoName,
 
-    #[snafu(display("object defines no spec"))]
-    ObjectHasNoSpec,
-
     #[snafu(display("object defines no namespace"))]
     ObjectHasNoNamespace,
 
@@ -222,21 +219,12 @@ pub enum Error {
     #[snafu(display("Failed to find information about file [{}] in product config", kind))]
     ProductConfigKindNotSpecified { kind: String },
 
-    #[snafu(display("Failed to find any nodes in cluster {obj_ref}",))]
-    MissingNodes {
-        source: stackable_operator::client::Error,
-        obj_ref: ObjectRef<v1alpha1::NifiCluster>,
-    },
-
     #[snafu(display("Failed to find service {obj_ref}"))]
     MissingService {
         source: stackable_operator::client::Error,
         obj_ref: ObjectRef<Service>,
     },
 
-    #[snafu(display("Failed to find an external port to use for proxy hosts"))]
-    ExternalPort,
-
     #[snafu(display("Could not build role service fqdn"))]
     NoRoleServiceFqdn,
 
@@ -534,8 +522,7 @@ pub async fn reconcile_nifi(
             // Since we cannot predict which of the addresses a user might decide to use we will simply
             // add all of them to the setting for now.
             // For more information see <https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#proxy_configuration>
-            // let proxy_hosts = get_proxy_hosts(client, nifi, &merged_config).await?;
-            let proxy_hosts = get_proxy_hosts(client, nifi).await?;
+            let proxy_hosts = get_proxy_hosts(client, nifi, &resolved_product_image).await?;
 
             let rg_configmap = build_node_rolegroup_config_map(
                 nifi,
@@ -1560,7 +1547,7 @@ async fn build_node_rolegroup_statefulset(
 async fn get_proxy_hosts(
     client: &Client,
     nifi: &v1alpha1::NifiCluster,
-    // merged_config: &NifiConfig,
+    resolved_product_image: &ResolvedProductImage,
 ) -> Result<String> {
     let host_header_check = nifi.spec.cluster_config.host_header_check.clone();
 
@@ -1576,27 +1563,24 @@ async fn get_proxy_hosts(
         return Ok("*".to_string());
     }
 
-    let node_role_service_fqdn = nifi
-        .node_role_service_fqdn(&client.kubernetes_cluster_info)
-        .context(NoRoleServiceFqdnSnafu)?;
-    let reporting_task_service_name = reporting_task::build_reporting_task_fqdn_service_name(
-        nifi,
-        &client.kubernetes_cluster_info,
-    )
-    .context(ReportingTaskSnafu)?;
-    let mut proxy_hosts_set = HashSet::from([
-        node_role_service_fqdn.clone(),
-        format!("{node_role_service_fqdn}:{HTTPS_PORT}"),
-        format!("{reporting_task_service_name}:{HTTPS_PORT}"),
+    // Address and port are injected from the listener volume during the prepare container
+    let mut proxy_hosts = HashSet::from([
+        "${env:LISTENER_DEFAULT_ADDRESS}:${env:LISTENER_DEFAULT_PORT_HTTPS}".to_string(),
     ]);
+    proxy_hosts.extend(host_header_check.additional_allowed_hosts);
 
-    proxy_hosts_set.extend(host_header_check.additional_allowed_hosts);
+    // Reporting task only exists for NiFi 1.x
+    if resolved_product_image.product_version.starts_with("1.") {
+        let reporting_task_service_name = reporting_task::build_reporting_task_fqdn_service_name(
+            nifi,
+            &client.kubernetes_cluster_info,
+        )
+        .context(ReportingTaskSnafu)?;
 
-    // Inject the address and port from the listener volume during the prepare container
-    proxy_hosts_set
-        .insert("${env:LISTENER_DEFAULT_ADDRESS}:${env:LISTENER_DEFAULT_PORT_HTTPS}".to_string());
+        proxy_hosts.insert(format!("{reporting_task_service_name}:{HTTPS_PORT}"));
+    }
 
-    let mut proxy_hosts = Vec::from_iter(proxy_hosts_set);
+    let mut proxy_hosts = Vec::from_iter(proxy_hosts);
     proxy_hosts.sort();
 
     Ok(proxy_hosts.join(","))

rust/operator-binary/src/crd/mod.rs

@@ -2,9 +2,8 @@ pub mod affinity;
 pub mod authentication;
 pub mod sensitive_properties;
 pub mod tls;
-pub mod utils;
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::BTreeMap;
 
 use affinity::get_affinity;
 use sensitive_properties::NifiSensitivePropertiesConfig;
@@ -46,7 +45,6 @@ use stackable_operator::{
     versioned::versioned,
 };
 use tls::NifiTls;
-use utils::PodRef;
 
 pub const APP_NAME: &str = "nifi";
 
@@ -84,9 +82,6 @@ pub enum Error {
 
     #[snafu(display("object has no nodes defined"))]
     NoNodesDefined,
-
-    #[snafu(display("listener podrefs could not be resolved"))]
-    ListenerPodRef { source: utils::Error },
 }
 
 #[versioned(version(name = "v1alpha1"))]
@@ -250,36 +245,6 @@ impl v1alpha1::NifiCluster {
         &self.spec.cluster_config.tls.server_secret_class
     }
 
-    /// List all pods expected to form the cluster
-    ///
-    /// We try to predict the pods here rather than looking at the current cluster state in order to
-    /// avoid instance churn.
-    pub fn pods(&self) -> Result<impl Iterator<Item = PodRef> + '_, Error> {
-        let ns = self.metadata.namespace.clone().context(NoNamespaceSnafu)?;
-        Ok(self
-            .spec
-            .nodes
-            .iter()
-            .flat_map(|role| &role.role_groups)
-            // Order rolegroups consistently, to avoid spurious downstream rewrites
-            .collect::<BTreeMap<_, _>>()
-            .into_iter()
-            .flat_map(move |(rolegroup_name, rolegroup)| {
-                let rolegroup_ref = self.node_rolegroup_ref(rolegroup_name);
-                let ns = ns.clone();
-                (0..rolegroup.replicas.unwrap_or(0)).map(move |i| PodRef {
-                    namespace: ns.clone(),
-                    role_group_service_name: rolegroup_ref.object_name(),
-                    pod_name: format!("{}-{}", rolegroup_ref.object_name(), i),
-                    ports: HashMap::from([
-                        (HTTPS_PORT_NAME.to_owned(), HTTPS_PORT),
-                        (METRICS_PORT_NAME.to_owned(), METRICS_PORT),
-                    ]),
-                    fqdn_override: None,
-                })
-            }))
-    }
-
     /// Retrieve and merge resource configs for role and role groups
     pub fn merged_config(&self, role: &NifiRole, role_group: &str) -> Result<NifiConfig, Error> {
         // Initialize the result with all default values as baseline

rust/operator-binary/src/crd/utils.rs

@@ -5,7 +5,7 @@ metadata:
   name: install-nifi
 timeout: 300
 commands:
-  - script: kubectl -n $NAMESPACE wait --for=condition=available=true nificlusters.nifi.stackable.tech/nifi --timeout 301s
+  - script: kubectl -n $NAMESPACE wait --for=condition=available=true nificlusters.nifi.stackable.tech/test-nifi --timeout 301s
 ---
 apiVersion: apps/v1
 kind: StatefulSet