diff --git a/rust/operator-binary/src/controller/build.rs b/rust/operator-binary/src/controller/build.rs index 8c713a32..7ba9d7c9 100644 --- a/rust/operator-binary/src/controller/build.rs +++ b/rust/operator-binary/src/controller/build.rs @@ -4,8 +4,23 @@ use std::str::FromStr; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::v2::types::{common::Port, operator::RoleGroupName}; +use crate::{ + controller::{ + KubernetesResources, ValidatedCluster, + build::resource::{ + config_map::build_rolegroup_config_map, + listener::{build_group_listener, group_listener_name}, + pdb::build_pdb, + service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, + statefulset::build_node_rolegroup_statefulset, + }, + }, + crd::NifiRole, +}; + pub mod git_sync; pub mod graceful_shutdown; pub mod jvm; @@ -27,3 +42,131 @@ pub const BALANCE_PORT: Port = Port(6243); // Filesystem paths shared by multiple builders. Single-consumer paths live in their builder. pub const NIFI_CONFIG_DIRECTORY: &str = "/stackable/nifi/conf"; pub const NIFI_PYTHON_WORKING_DIRECTORY: &str = "/nifi-python-working-directory"; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("NifiCluster has no nodes role defined"))] + NoNodesDefined, + + #[snafu(display("failed to build ConfigMap for role group {role_group}"))] + ConfigMap { + source: resource::config_map::Error, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build StatefulSet for role group {role_group}"))] + StatefulSet { + source: resource::statefulset::Error, + role_group: RoleGroupName, + }, +} + +/// Builds every Kubernetes resource for the given validated cluster. +/// +/// Does not need a Kubernetes client: every reference to another Kubernetes resource is already +/// dereferenced and validated by this point, so the errors returned here are resource-assembly +/// failures only. +/// +/// `service_account_name` is the name of the RBAC `ServiceAccount` the role-group Pods run under +/// (RBAC resources are built and applied separately, in the reconcile step). +pub fn build( + cluster: &ValidatedCluster, + service_account_name: &str, +) -> Result { + let mut stateful_sets = vec![]; + let mut services = vec![]; + let mut listeners = vec![]; + let mut config_maps = vec![]; + let mut pod_disruption_budgets = vec![]; + + // NiFi has a single role (`node`), which must always be present. + let nifi_role = NifiRole::Node; + let node_role_group_configs = cluster + .role_group_configs + .get(&nifi_role) + .context(NoNodesDefinedSnafu)?; + + // Role-level resources (one per role): the PodDisruptionBudget and the group Listener. + let role_config = &cluster.role_config; + if let Some(pdb) = build_pdb(&role_config.pdb, cluster, &nifi_role) { + pod_disruption_budgets.push(pdb); + } + listeners.push(build_group_listener( + cluster, + role_config.listener_class.clone(), + group_listener_name(cluster, &nifi_role.to_string()), + )); + + for (role_group_name, rg) in node_role_group_configs { + services.push(build_rolegroup_headless_service(cluster, role_group_name)); + services.push(build_rolegroup_metrics_service(cluster, role_group_name)); + + config_maps.push( + build_rolegroup_config_map(cluster, role_group_name, rg).context(ConfigMapSnafu { + role_group: role_group_name.clone(), + })?, + ); + + let effective_replicas = rg.replicas.map(i32::from); + stateful_sets.push( + build_node_rolegroup_statefulset( + cluster, + role_group_name, + rg, + effective_replicas, + service_account_name, + ) + .context(StatefulSetSnafu { + role_group: role_group_name.clone(), + })?, + ); + } + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + }) +} + +#[cfg(test)] +mod tests { + use stackable_operator::kube::Resource; + + use super::{build, properties::test_support::minimal_validated_cluster}; + + fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { + let mut names: Vec<&str> = resources + .iter() + .filter_map(|resource| resource.meta().name.as_deref()) + .collect(); + names.sort(); + names + } + + #[test] + fn build_produces_expected_resources() { + let cluster = minimal_validated_cluster(); + let resources = build(&cluster, "simple-nifi-serviceaccount").expect("build succeeds"); + + // The minimal fixture has a single `default` role group for the `node` role. + assert_eq!( + sorted_names(&resources.stateful_sets), + ["simple-nifi-node-default"] + ); + assert_eq!( + sorted_names(&resources.config_maps), + ["simple-nifi-node-default"] + ); + // One headless and one metrics Service per role group. + assert_eq!(resources.services.len(), 2); + // One group Listener and one PDB for the single `node` role. + assert_eq!(sorted_names(&resources.listeners), ["simple-nifi-node"]); + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["simple-nifi-node"] + ); + } +} diff --git a/rust/operator-binary/src/controller/build/properties.rs b/rust/operator-binary/src/controller/build/properties.rs index a03c6c5b..58c4ba0c 100644 --- a/rust/operator-binary/src/controller/build/properties.rs +++ b/rust/operator-binary/src/controller/build/properties.rs @@ -81,7 +81,7 @@ pub(crate) mod test_support { use std::str::FromStr as _; use stackable_operator::{ - commons::product_image_selection::ResolvedProductImage, + commons::{networking::DomainName, product_image_selection::ResolvedProductImage}, crd::authentication::r#static::v1alpha1::{ AuthenticationProvider as StaticAuthProvider, UserCredentialsSecretRef, }, @@ -168,6 +168,7 @@ pub(crate) mod test_support { ValidatedCluster::new( name, namespace, + DomainName::from_str("cluster.local").expect("valid cluster domain"), uid, image, product_version, diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 64bd3b37..8c01c35f 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -32,7 +32,7 @@ use stackable_operator::{ self, framework::{create_vector_shutdown_file_command, remove_vector_shutdown_file_command}, }, - utils::{COMMON_BASH_TRAP_FUNCTIONS, cluster_info::KubernetesClusterInfo}, + utils::COMMON_BASH_TRAP_FUNCTIONS, v2::{ builder::pod::container::{EnvVarSet, new_container_builder}, product_logging::framework::{ @@ -164,10 +164,8 @@ pub(crate) const LISTENER_DEFAULT_PORT_HTTPS_ENV: &str = "LISTENER_DEFAULT_PORT_ /// /// The [`Pod`](`stackable_operator::k8s_openapi::api::core::v1::Pod`)s are accessible through the /// corresponding [`stackable_operator::k8s_openapi::api::core::v1::Service`] (from `build_rolegroup_headless_service`). -#[allow(clippy::too_many_arguments)] -pub(crate) async fn build_node_rolegroup_statefulset( +pub(crate) fn build_node_rolegroup_statefulset( cluster: &ValidatedCluster, - cluster_info: &KubernetesClusterInfo, role_group_name: &RoleGroupName, rg: &NifiRoleGroupConfig, effective_replicas: Option, @@ -251,7 +249,7 @@ pub(crate) async fn build_node_rolegroup_statefulset( "$POD_NAME.{service_name}.{namespace}.svc.{cluster_domain}", service_name = resource_names.headless_service_name(), namespace = cluster.namespace, - cluster_domain = cluster_info.cluster_domain, + cluster_domain = cluster.cluster_domain, ); let sensitive_key_secret = &cluster.cluster_config.sensitive_properties.key_secret; diff --git a/rust/operator-binary/src/controller/dereference.rs b/rust/operator-binary/src/controller/dereference.rs index a5d31cb6..5eadb6ed 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -6,6 +6,7 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ client::Client, + commons::networking::DomainName, v2::{ controller_utils::{self, get_namespace}, types::kubernetes::NamespaceName, @@ -38,6 +39,9 @@ type Result = std::result::Result; pub struct DereferencedObjects { /// The namespace of the [`v1alpha1::NifiCluster`], parsed once here and reused everywhere. pub namespace: NamespaceName, + /// The Kubernetes cluster domain, captured from the client here so the build step needs no + /// client to assemble in-cluster DNS names. + pub cluster_domain: DomainName, pub authentication_classes: DereferencedAuthenticationClasses, pub authorization: DereferencedAuthorization, } @@ -63,6 +67,7 @@ pub async fn dereference( Ok(DereferencedObjects { namespace, + cluster_domain: client.kubernetes_cluster_info.cluster_domain.clone(), authentication_classes, authorization, }) diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index f9106741..4098ad17 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -7,11 +7,19 @@ use std::{collections::BTreeMap, str::FromStr as _}; use stackable_operator::{ commons::{ affinity::StackableAffinity, + networking::DomainName, product_image_selection::ResolvedProductImage, resources::{NoRuntimeLimits, Resources}, }, - crd::git_sync, - k8s_openapi::{api::core::v1::Volume, apimachinery::pkg::apis::meta::v1::ObjectMeta}, + crd::{git_sync, listener}, + k8s_openapi::{ + api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service, Volume}, + policy::v1::PodDisruptionBudget, + }, + apimachinery::pkg::apis::meta::v1::ObjectMeta, + }, kube::Resource, kvp::Labels, shared::time::Duration, @@ -47,6 +55,15 @@ pub(crate) mod build; pub(crate) mod dereference; pub(crate) mod validate; +/// Every Kubernetes resource produced by the [`build`] step. +pub struct KubernetesResources { + pub stateful_sets: Vec, + pub services: Vec, + pub listeners: Vec, + pub config_maps: Vec, + pub pod_disruption_budgets: Vec, +} + /// A validated, merged (default <- role <- role-group) NiFi rolegroup config. pub type NifiRoleGroupConfig = RoleGroupConfig; @@ -140,6 +157,9 @@ pub struct ValidatedCluster { pub name: ClusterName, /// The namespace of the NifiCluster, parsed once in the dereference step and reused everywhere. pub namespace: NamespaceName, + /// The Kubernetes cluster domain, captured from the client in the dereference step so the + /// build step needs no client to assemble in-cluster DNS names. + pub cluster_domain: DomainName, /// The UID of the NifiCluster, used to build OwnerReferences downstream. pub uid: Uid, /// The product image. @@ -198,6 +218,7 @@ impl ValidatedCluster { pub fn new( name: ClusterName, namespace: NamespaceName, + cluster_domain: DomainName, uid: Uid, image: ResolvedProductImage, product_version: ProductVersion, @@ -216,6 +237,7 @@ impl ValidatedCluster { metadata, name, namespace, + cluster_domain, uid, image, product_version, diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index a29ed629..c5d7950a 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -156,6 +156,7 @@ pub fn validate( .context(NoNodesDefinedSnafu)?; let namespace = dereferenced_objects.namespace.clone(); + let cluster_domain = dereferenced_objects.cluster_domain.clone(); let uid = get_uid(nifi).context(GetUidSnafu)?; // `app_version_label_value` is constructed to be a valid label value, so it is always a valid @@ -166,6 +167,7 @@ pub fn validate( Ok(ValidatedCluster::new( name, namespace, + cluster_domain, uid, image, product_version, diff --git a/rust/operator-binary/src/nifi_controller.rs b/rust/operator-binary/src/nifi_controller.rs index b10e9013..0cee34c5 100644 --- a/rust/operator-binary/src/nifi_controller.rs +++ b/rust/operator-binary/src/nifi_controller.rs @@ -3,7 +3,7 @@ use std::{str::FromStr, sync::Arc}; use const_format::concatcp; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, client::Client, @@ -20,27 +20,14 @@ use stackable_operator::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, - v2::{ - cluster_resources::cluster_resources_new, - types::operator::{ProductVersion, RoleGroupName}, - }, + v2::{cluster_resources::cluster_resources_new, types::operator::ProductVersion}, }; use strum::{EnumDiscriminants, IntoStaticStr}; -use tracing::Instrument; use crate::{ OPERATOR_NAME, - controller::{ - build, - build::resource::{ - listener::{build_group_listener, group_listener_name}, - pdb::build_pdb, - service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, - statefulset::build_node_rolegroup_statefulset, - }, - controller_name, dereference, operator_name, product_name, validate, - }, - crd::{APP_NAME, NifiRole, NifiStatus, v1alpha1}, + controller::{build, controller_name, dereference, operator_name, product_name, validate}, + crd::{APP_NAME, NifiStatus, v1alpha1}, security::{ authentication::NifiAuthenticationConfig, check_or_generate_oidc_admin_password, check_or_generate_sensitive_key, @@ -80,37 +67,12 @@ pub enum Error { source: stackable_operator::client::Error, }, - #[snafu(display("failed to apply Service for {}", rolegroup))] - ApplyRoleGroupService { - source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to build rolegroup ConfigMap for {}", rolegroup))] - BuildRoleGroupConfigMap { - source: build::resource::config_map::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to build StatefulSet for {}", rolegroup))] - BuildStatefulSet { - source: crate::controller::build::resource::statefulset::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("object has no nodes defined"))] - NoNodesDefined, - - #[snafu(display("failed to apply ConfigMap for {}", rolegroup))] - ApplyRoleGroupConfig { - source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, - }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to apply StatefulSet for {}", rolegroup))] - ApplyRoleGroupStatefulSet { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, }, #[snafu(display("failed to patch service account"))] @@ -128,11 +90,6 @@ pub enum Error { source: stackable_operator::commons::rbac::Error, }, - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to get required labels"))] GetRequiredLabels { source: @@ -141,11 +98,6 @@ pub enum Error { #[snafu(display("security failure"))] Security { source: crate::security::Error }, - - #[snafu(display("failed to apply group listener"))] - ApplyGroupListener { - source: stackable_operator::cluster_resources::Error, - }, } type Result = std::result::Result; @@ -231,105 +183,46 @@ pub async fn reconcile_nifi( .await .context(ApplyRoleBindingSnafu)?; + let resources = + build::build(&validated_cluster, &rbac_sa.name_any()).context(BuildResourcesSnafu)?; + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - let nifi_role = NifiRole::Node; - let node_role_group_configs = validated_cluster - .role_group_configs - .get(&nifi_role) - .context(NoNodesDefinedSnafu)?; - for (role_group_name, rg) in node_role_group_configs.iter() { - let rg_span = tracing::info_span!("rolegroup_span", rolegroup = role_group_name.as_ref()); - async { - tracing::debug!("Processing rolegroup {role_group_name}"); - - let rg_headless_service = - build_rolegroup_headless_service(&validated_cluster, role_group_name); - - let rg_configmap = build::resource::config_map::build_rolegroup_config_map( - &validated_cluster, - role_group_name, - rg, - ) - .context(BuildRoleGroupConfigMapSnafu { - rolegroup: role_group_name.clone(), - })?; - - let effective_replicas = rg.replicas.map(i32::from); - - let rg_statefulset = build_node_rolegroup_statefulset( - &validated_cluster, - &client.kubernetes_cluster_info, - role_group_name, - rg, - effective_replicas, - &rbac_sa.name_any(), - ) + // Apply order: everything before StatefulSets, StatefulSets last. A StatefulSet must be applied + // after all ConfigMaps and Secrets it mounts, otherwise the Pods restart unnecessarily. + // See https://github.com/stackabletech/commons-operator/issues/111 for details. + for service in resources.services { + cluster_resources + .add(client, service) .await - .with_context(|_| BuildStatefulSetSnafu { - rolegroup: role_group_name.clone(), - })?; - - let rg_metrics_service = - build_rolegroup_metrics_service(&validated_cluster, role_group_name); - - cluster_resources - .add(client, rg_metrics_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - rolegroup: role_group_name.clone(), - })?; - - cluster_resources - .add(client, rg_headless_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - rolegroup: role_group_name.clone(), - })?; - - cluster_resources - .add(client, rg_configmap) - .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - rolegroup: role_group_name.clone(), - })?; - - // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts - // to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - ss_cond_builder.add( - cluster_resources - .add(client, rg_statefulset) - .await - .with_context(|_| ApplyRoleGroupStatefulSetSnafu { - rolegroup: role_group_name.clone(), - })?, - ); - - Ok(()) - } - .instrument(rg_span) - .await? + .context(ApplyResourceSnafu)?; } - - let role_config = &validated_cluster.role_config; - if let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, &nifi_role) { + for listener in resources.listeners { + cluster_resources + .add(client, listener) + .await + .context(ApplyResourceSnafu)?; + } + for config_map in resources.config_maps { + cluster_resources + .add(client, config_map) + .await + .context(ApplyResourceSnafu)?; + } + for pdb in resources.pod_disruption_budgets { cluster_resources .add(client, pdb) .await - .context(ApplyPdbSnafu)?; + .context(ApplyResourceSnafu)?; + } + for stateful_set in resources.stateful_sets { + ss_cond_builder.add( + cluster_resources + .add(client, stateful_set) + .await + .context(ApplyResourceSnafu)?, + ); } - - let role_group_listener = build_group_listener( - &validated_cluster, - role_config.listener_class.clone(), - group_listener_name(&validated_cluster, &nifi_role.to_string()), - ); - - cluster_resources - .add(client, role_group_listener) - .await - .context(ApplyGroupListenerSnafu)?; // Remove any orphaned resources that still exist in k8s, but have not been added to // the cluster resources during the reconciliation