diff --git a/CHANGELOG.md b/CHANGELOG.md index 2845c927..5bf76d8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,10 +13,13 @@ All notable changes to this project will be documented in this file. functions and carry the full set of recommended labels ([#1060]). - BREAKING: The `servers` role is now required by the CRD. Previously a ZookeeperCluster without it was accepted by the API server but failed reconciliation ([#1060]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps for the `zk_controller` and `znode_controller` ([#1069]). [#1053]: https://github.com/stackabletech/zookeeper-operator/pull/1053 [#1060]: https://github.com/stackabletech/zookeeper-operator/pull/1060 [#1063]: https://github.com/stackabletech/zookeeper-operator/pull/1063 +[#1069]: https://github.com/stackabletech/zookeeper-operator/pull/1069 ## [26.7.0] - 2026-07-21 diff --git a/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml b/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml index 651db149..8ed7ed5b 100644 --- a/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml @@ -118,8 +118,9 @@ rules: verbs: - create - patch - # Listener created per role group for external access. Applied via SSA and tracked for - # orphan cleanup. + # Listener created per role for external access. Applied via SSA and tracked for + # orphan cleanup. Watched as well, because the discovery ConfigMap advertises the addresses + # that the listener operator publishes on it. - apiGroups: - listeners.stackable.tech resources: @@ -130,6 +131,7 @@ rules: - get - list - patch + - watch # Primary CRD: watched and read during reconciliation. - apiGroups: - {{ include "operator.name" . }}.stackable.tech diff --git a/rust/operator-binary/src/discovery.rs b/rust/operator-binary/src/discovery.rs new file mode 100644 index 00000000..a8e1c0a6 --- /dev/null +++ b/rust/operator-binary/src/discovery.rs @@ -0,0 +1,151 @@ +//! Builders for the discovery ConfigMaps, which advertise how to connect to a ZooKeeper ensemble. +//! +//! Shared by the build steps of both controllers: the ZookeeperCluster controller publishes the +//! whole ensemble, the ZookeeperZnode controller publishes the same ensemble narrowed to a chroot. + +use std::str::FromStr; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder}, + k8s_openapi::api::core::v1::ConfigMap, + kube::Resource, + v2::{ + HasName, HasUid, NameIsValidLabelValue, + builder::meta::ownerreference_from_resource, + kvp::label::recommended_labels, + types::operator::{ControllerName, ProductVersion, RoleGroupName}, + }, +}; + +use crate::{ + crd::{ZookeeperRole, security::ZookeeperSecurity}, + listener_addresses::ListenerAddresses, + zk_controller::validate::{ValidatedCluster, operator_name, product_name}, + znode_controller::validate::ValidatedZnode, +}; + +// Placeholder role-group name used for the recommended labels of the role-level discovery +// `ConfigMap` (which is not tied to a single role group). +stackable_operator::constant!(PLACEHOLDER_DISCOVERY_ROLE_GROUP: RoleGroupName = "discovery"); + +type Result = std::result::Result; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("chroot path {} was relative (must be absolute)", chroot))] + RelativeChroot { chroot: String }, + + #[snafu(display("failed to build ConfigMap"))] + BuildConfigMap { + source: stackable_operator::builder::configmap::Error, + }, +} + +/// Build the discovery [`ConfigMap`] for the cluster controller from the +/// [`ValidatedCluster`]. +/// +/// The ConfigMap is owned by, and placed in the namespace of, the cluster. The image and security +/// settings are taken from the [`ValidatedCluster`] rather than being passed in separately. +pub fn build_discovery_configmap( + validated_cluster: &ValidatedCluster, + controller_name: &str, + listener_addresses: &ListenerAddresses, +) -> Result { + build_discovery_configmap_for_owner( + validated_cluster, + &validated_cluster.namespace, + controller_name, + &validated_cluster.product_version, + listener_addresses, + None, + &validated_cluster.cluster_config.zookeeper_security, + ) +} + +/// Build the discovery [`ConfigMap`] for the znode controller. +/// +/// The ConfigMap is owned by, and placed in the namespace of, the +/// [`ValidatedZnode`]. The product version and `zookeeper_security` originate from the referenced +/// cluster (via the validated znode), while `chroot` isolates the znode within the shared ZooKeeper +/// ensemble. +pub fn build_znode_discovery_configmap( + validated_znode: &ValidatedZnode, + controller_name: &str, + listener_addresses: &ListenerAddresses, + chroot: &str, +) -> Result { + build_discovery_configmap_for_owner( + validated_znode, + &validated_znode.namespace, + controller_name, + &validated_znode.product_version, + listener_addresses, + Some(chroot), + &validated_znode.zookeeper_security, + ) +} + +/// Build a discovery [`ConfigMap`] containing ZooKeeper connection details from a +/// [`listener::v1alpha1::Listener`]. +/// +/// `owner` owns the ConfigMap (the [`ZookeeperCluster`](crate::crd::v1alpha1::ZookeeperCluster) for the cluster +/// controller, or the [`ZookeeperZnode`](crate::crd::v1alpha1::ZookeeperZnode) for the znode controller) and +/// `namespace` is where the ConfigMap is placed. +fn build_discovery_configmap_for_owner( + owner: &(impl Resource + HasName + HasUid + NameIsValidLabelValue), + namespace: impl Into, + controller_name: &str, + product_version: &ProductVersion, + listener_addresses: &ListenerAddresses, + chroot: Option<&str>, + zookeeper_security: &ZookeeperSecurity, +) -> Result { + let name = owner.to_name(); + + // The discovery ConfigMap is a role-level resource of the `server` role, conventionally + // labelled with the `discovery` role group. The controller name differs between the cluster and + // znode controllers, so it is passed in and validated into the type-safe newtype here. + let controller_name = ControllerName::from_str(controller_name) + .expect("the controller name is a valid label value"); + let role_group_name = PLACEHOLDER_DISCOVERY_ROLE_GROUP.clone(); + + // Write a connection string of the format that Java ZooKeeper client expects: + // "{host1}:{port1},{host2:port2},.../{chroot}" + // See https://zookeeper.apache.org/doc/current/apidocs/zookeeper-server/org/apache/zookeeper/ZooKeeper.html#ZooKeeper-java.lang.String-int-org.apache.zookeeper.Watcher- + let listener_addresses = listener_addresses.to_connection_string(); + let mut conn_str = listener_addresses.clone(); + if let Some(chroot) = chroot { + if !chroot.starts_with('/') { + return RelativeChrootSnafu { chroot }.fail(); + } + conn_str.push_str(chroot); + } + ConfigMapBuilder::new() + .metadata( + ObjectMetaBuilder::new() + .name(name) + .namespace(namespace) + .ownerreference(ownerreference_from_resource(owner, None, Some(true))) + .with_labels(recommended_labels( + owner, + &product_name(), + product_version, + &operator_name(), + &controller_name, + &ZookeeperRole::Server.into(), + &role_group_name, + )) + .build(), + ) + .add_data("ZOOKEEPER", conn_str) + // Some clients don't support ZooKeeper's merged `hosts/chroot` format, so export them separately for these clients + .add_data("ZOOKEEPER_HOSTS", listener_addresses) + .add_data( + "ZOOKEEPER_CLIENT_PORT", + zookeeper_security.client_port().to_string(), + ) + .add_data("ZOOKEEPER_CHROOT", chroot.unwrap_or("/")) + .build() + .context(BuildConfigMapSnafu) +} diff --git a/rust/operator-binary/src/listener_addresses.rs b/rust/operator-binary/src/listener_addresses.rs new file mode 100644 index 00000000..4eaa1e0d --- /dev/null +++ b/rust/operator-binary/src/listener_addresses.rs @@ -0,0 +1,164 @@ +//! Reading client connection addresses from a [`Listener`](listener::v1alpha1::Listener). +//! +//! Shared by both controllers, which turn the dereferenced ZooKeeper role Listener into the +//! addresses advertised by their discovery ConfigMaps. + +use std::{collections::BTreeSet, num::TryFromIntError}; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{crd::listener, kube::runtime::reflector::ObjectRef}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("{listener} does not have a port with the name {port_name:?}"))] + PortNotFound { + port_name: String, + listener: ObjectRef, + }, + + #[snafu(display("expected an unsigned 16-bit port, got {port_number}"))] + InvalidPort { + source: TryFromIntError, + port_number: i32, + }, +} + +type Result = std::result::Result; + +/// The address and port pairs published by a [`Listener`](listener::v1alpha1::Listener) for a +/// single named port, sorted and deduplicated. +/// +/// An address is a hostname or IP address of a node, a cluster IP or an external load balancer, +/// depending on the Service type behind the Listener. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ListenerAddresses(BTreeSet<(String, u16)>); + +impl ListenerAddresses { + /// Renders the addresses as the comma separated `host1:port1,host2:port2` list that ZooKeeper + /// clients expect. + pub fn to_connection_string(&self) -> String { + self.0 + .iter() + .map(|(address, port)| format!("{address}:{port}")) + .collect::>() + .join(",") + } +} + +/// Reads the addresses that `listener` publishes for `port_name`. +/// +/// Returns `Ok(None)` while the Listener carries no ingress addresses at all, which is the normal +/// state between creating the Listener and the listener operator publishing its addresses. A +/// Listener that does publish addresses, but none for `port_name`, is an error. +// TODO (@NickLarsenNZ): Move this to stackable-operator, so it can be used as +// listener.addresses_for_port(port_name) +pub fn listener_addresses( + listener: &listener::v1alpha1::Listener, + port_name: &str, +) -> Result> { + let Some(ingress_addresses) = listener + .status + .as_ref() + .and_then(|listener_status| listener_status.ingress_addresses.as_ref()) + else { + return Ok(None); + }; + + let address_port_pairs = ingress_addresses + .iter() + // Filter the addresses that have the port we are interested in (they likely all have it though) + .filter_map(|listener_ingress| { + Some(listener_ingress.address.clone()).zip(listener_ingress.ports.get(port_name)) + }) + // Convert the port from i32 to u16 + .map(|(listener_address, &port_number)| { + let port_number: u16 = port_number + .try_into() + .context(InvalidPortSnafu { port_number })?; + Ok((listener_address, port_number)) + }) + .collect::, _>>()?; + + match address_port_pairs.is_empty() { + true => PortNotFoundSnafu { + port_name, + listener, + } + .fail(), + false => Ok(Some(ListenerAddresses(address_port_pairs))), + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use stackable_operator::{ + crd::listener::v1alpha1::{ + AddressType, Listener, ListenerIngress, ListenerSpec, ListenerStatus, + }, + k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + + use super::*; + use crate::crd::ZOOKEEPER_SERVER_PORT_NAME; + + fn listener(ingress_addresses: Option>) -> Listener { + Listener { + metadata: ObjectMeta { + name: Some("test-listener".to_owned()), + ..ObjectMeta::default() + }, + spec: ListenerSpec::default(), + status: Some(ListenerStatus { + service_name: None, + ingress_addresses, + node_ports: None, + }), + } + } + + fn ingress(port: i32) -> ListenerIngress { + ListenerIngress { + address: "node-0".to_owned(), + address_type: AddressType::Hostname, + ports: BTreeMap::from([(ZOOKEEPER_SERVER_PORT_NAME.to_owned(), port)]), + } + } + + #[test] + fn listener_addresses_returns_host_port_pairs() { + let listener = listener(Some(vec![ingress(2181)])); + let addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) + .expect("addresses") + .expect("the listener publishes addresses"); + assert_eq!(addresses.to_connection_string(), "node-0:2181"); + } + + #[test] + fn listener_addresses_without_ingress_is_not_ready_yet() { + assert_eq!( + listener_addresses(&listener(None), ZOOKEEPER_SERVER_PORT_NAME).expect("addresses"), + None + ); + } + + #[test] + fn listener_addresses_missing_port_name_is_error() { + let listener = listener(Some(vec![ingress(2181)])); + assert!(matches!( + listener_addresses(&listener, "does-not-exist"), + Err(Error::PortNotFound { .. }) + )); + } + + #[test] + fn listener_addresses_port_out_of_u16_range_is_error() { + // A port number that does not fit into a u16 must be rejected. + let listener = listener(Some(vec![ingress(70_000)])); + assert!(matches!( + listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME), + Err(Error::InvalidPort { .. }) + )); + } +} diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 1b9c6067..c8babfeb 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -13,6 +13,7 @@ use futures::{FutureExt, StreamExt, TryFutureExt}; use stackable_operator::{ YamlSchema, cli::{Command, RunArguments}, + crd::listener::v1alpha1::Listener, eos::EndOfSupportChecker, k8s_openapi::api::{ apps::v1::StatefulSet, @@ -40,6 +41,8 @@ use crate::{ }; pub mod crd; +mod discovery; +mod listener_addresses; mod webhooks; mod zk_controller; mod znode_controller; @@ -139,6 +142,13 @@ async fn main() -> anyhow::Result<()> { watch_namespace.get_api::>(&client), watcher::Config::default(), ) + // The discovery ConfigMap advertises the addresses that the listener operator + // publishes on the role Listener, so a reconciliation must run once they appear + // or change. + .owns( + watch_namespace.get_api::>(&client), + watcher::Config::default(), + ) .graceful_shutdown_on(sigterm_watcher.handle()) .run( zk_controller::reconcile_zk, diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index 07296bae..9a116ad5 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -1,9 +1,12 @@ //! Ensures that `Pod`s are configured and running for each [`v1alpha1::ZookeeperCluster`] -use std::{hash::Hasher, str::FromStr, sync::Arc}; +//! +//! This is the controller driver: it runs the +//! `dereference -> validate -> build -> apply -> update_status` pipeline, with each step living +//! in its own submodule. +use std::{marker::PhantomData, sync::Arc}; use const_format::concatcp; -use fnv::FnvHasher; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, @@ -21,25 +24,19 @@ use stackable_operator::{ }, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, - v2::{cluster_resources::cluster_resources_new, types::operator::ControllerName}, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ OPERATOR_NAME, ObjectRef, crd::v1alpha1, - zk_controller::{ - build::resource::discovery, - validate::{operator_name, product_name}, - }, + zk_controller::{apply::Applier, update_status::update_status}, }; +pub(crate) mod apply; pub(crate) mod build; mod dereference; +mod update_status; pub(crate) mod validate; pub const ZK_CONTROLLER_NAME: &str = "zookeepercluster"; @@ -71,43 +68,11 @@ pub enum Error { #[snafu(display("failed to build the Kubernetes resources"))] BuildResources { source: build::Error }, - #[snafu(display("failed to apply Kubernetes resource"))] - ApplyResource { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("object is missing metadata to build owner reference"))] - ObjectMissingMetadataForOwnerRef { - source: stackable_operator::builder::meta::Error, - }, - - #[snafu(display( - "no role Listener was applied; the discovery ConfigMap is derived from the applied role Listener" - ))] - NoRoleListener, + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, - #[snafu(display("failed to build discovery ConfigMap"))] - BuildDiscoveryConfig { source: discovery::Error }, - - #[snafu(display("failed to apply discovery ConfigMap"))] - ApplyDiscoveryConfig { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphans { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to build object meta data"))] - ObjectMeta { - source: stackable_operator::builder::meta::Error, - }, + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, } impl ReconcilerError for Error { @@ -121,29 +86,36 @@ impl ReconcilerError for Error { Error::Dereference { .. } => None, Error::ValidateCluster { .. } => None, Error::BuildResources { .. } => None, - Error::ApplyResource { .. } => None, - Error::ObjectMissingMetadataForOwnerRef { .. } => None, - Error::NoRoleListener => None, - Error::BuildDiscoveryConfig { .. } => None, - Error::ApplyDiscoveryConfig { .. } => None, - Error::ApplyStatus { .. } => None, - Error::DeleteOrphans { .. } => None, - Error::ObjectMeta { .. } => None, + Error::ApplyResources { .. } => None, + Error::UpdateStatus { .. } => None, } } } +/// Marker for prepared Kubernetes resources which are not applied yet. +pub struct Prepared; + +/// Marker for applied Kubernetes resources. +pub struct Applied; + /// Every Kubernetes resource produced by the client-free [`build()`](build::build) step. /// -/// The discovery `ConfigMap` is deliberately absent — see [`build()`](build::build). -pub struct KubernetesResources { +/// `T` is a marker that indicates if these resources are only [`Prepared`] or already [`Applied`]. +/// The marker is useful e.g. to ensure that the cluster status is updated based on the applied +/// resources. +pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, pub listeners: Vec, pub config_maps: Vec, + /// The discovery `ConfigMap`, which is only built once the role Listener publishes its + /// addresses (see [`build()`](build::build)). It is kept apart from the role group + /// `config_maps` because the cluster status carries a hash of it. + pub maybe_discovery_config_map: Option, pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } pub async fn reconcile_zk( @@ -167,116 +139,25 @@ pub async fn reconcile_zk( validate::validate(zk, &dereferenced_objects, &ctx.operator_environment) .context(ValidateClusterSnafu)?; - // Names are derived from compile-time constants. - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &ControllerName::from_str(ZK_CONTROLLER_NAME) - .expect("ZK_CONTROLLER_NAME should be a valid controller name"), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&validated_cluster.cluster_operation), - &validated_cluster.object_overrides, - ); - + // build (no client required) let resources = build::build(&validated_cluster, &client.kubernetes_cluster_info) .context(BuildResourcesSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - - for service_account in resources.service_accounts { - cluster_resources - .add(client, service_account) - .await - .context(ApplyResourceSnafu)?; - } - for role_binding in resources.role_bindings { - cluster_resources - .add(client, role_binding) - .await - .context(ApplyResourceSnafu)?; - } - - for service in resources.services { - cluster_resources - .add(client, service) - .await - .context(ApplyResourceSnafu)?; - } - - // ZooKeeper has a single role Listener; the applied object feeds the discovery ConfigMap. - let mut applied_role_listener: Option = None; - for listener in resources.listeners { - applied_role_listener = Some( - cluster_resources - .add(client, listener) - .await - .context(ApplyResourceSnafu)?, - ); - } - let role_listener = applied_role_listener.context(NoRoleListenerSnafu)?; - - 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(ApplyResourceSnafu)?; - } - - // 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. - for statefulset in resources.stateful_sets { - ss_cond_builder.add( - cluster_resources - .add(client, statefulset) - .await - .context(ApplyResourceSnafu)?, - ); - } - - // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. - // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. - let mut discovery_hash = FnvHasher::with_key(0); - - let discovery_cm = - discovery::build_discovery_configmap(&validated_cluster, ZK_CONTROLLER_NAME, role_listener) - .context(BuildDiscoveryConfigSnafu)?; - - let discovery_cm = cluster_resources - .add(client, discovery_cm) - .await - .context(ApplyDiscoveryConfigSnafu)?; - if let Some(generation) = discovery_cm.metadata.resource_version { - discovery_hash.write(generation.as_bytes()) - } - - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&zk.spec.cluster_operation); - - let status = v1alpha1::ZookeeperClusterStatus { - // Serialize as a string to discourage users from trying to parse the value, - // and to keep things flexible if we end up changing the hasher at some point. - discovery_hash: Some(discovery_hash.finish().to_string()), - conditions: compute_conditions(zk, &[&ss_cond_builder, &cluster_operation_cond_builder]), - }; + // apply (client required) + let applied = Applier::new( + client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&validated_cluster.cluster_operation), + &validated_cluster.object_overrides, + ) + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphansSnafu)?; - client - .apply_patch_status(OPERATOR_NAME, zk, &status) + // update_status (client required) + update_status(client, zk, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(controller::Action::await_change()) } @@ -361,6 +242,7 @@ pub(crate) mod test_support { zk, &DereferencedObjects { authentication_classes: DereferencedAuthenticationClasses::new_for_tests(), + maybe_role_listener: None, }, &operator_environment(), ) diff --git a/rust/operator-binary/src/zk_controller/apply.rs b/rust/operator-binary/src/zk_controller/apply.rs new file mode 100644 index 00000000..94d34d0c --- /dev/null +++ b/rust/operator-binary/src/zk_controller/apply.rs @@ -0,0 +1,141 @@ +//! The apply step in the ZookeeperCluster controller. + +use std::{marker::PhantomData, str::FromStr}; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + v2::{cluster_resources::cluster_resources_new, types::operator::ControllerName}, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::zk_controller::{ + Applied, KubernetesResources, Prepared, ZK_CONTROLLER_NAME, + validate::{ValidatedCluster, operator_name, product_name}, +}; + +#[derive(Snafu, Debug, EnumDiscriminants)] +#[strum_discriminants(derive(IntoStaticStr))] +pub enum Error { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { + source: stackable_operator::cluster_resources::Error, + }, + + #[snafu(display("failed to delete orphaned resources"))] + DeleteOrphanedResources { + source: stackable_operator::cluster_resources::Error, + }, +} + +type Result = std::result::Result; + +/// Applier for the Kubernetes resource specifications produced by this controller. +/// +/// The implementation is not tied to this controller and could theoretically be moved to +/// stackable_operator if [`KubernetesResources`] would contain all possible resource types. +pub struct Applier<'a> { + client: &'a Client, + cluster_resources: ClusterResources<'a>, +} + +impl<'a> Applier<'a> { + pub fn new( + client: &'a Client, + cluster: &ValidatedCluster, + apply_strategy: ClusterResourceApplyStrategy, + object_overrides: &'a ObjectOverrides, + ) -> Applier<'a> { + // Names are derived from compile-time constants. + let cluster_resources = cluster_resources_new( + &product_name(), + &operator_name(), + &ControllerName::from_str(ZK_CONTROLLER_NAME) + .expect("ZK_CONTROLLER_NAME should be a valid controller name"), + &cluster.name, + &cluster.namespace, + &cluster.uid, + apply_strategy, + object_overrides, + ); + + Applier { + client, + cluster_resources, + } + } + + /// Applies the given Kubernetes resources and marks them as applied. + pub async fn apply( + mut self, + resources: KubernetesResources, + ) -> Result> { + // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to + // compile here instead of silently never being applied. + let KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + maybe_discovery_config_map, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: _, + } = resources; + + // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret must exist first, + // else Pods restart, see commons-operator#111). The ServiceAccount comes first because the + // Pods reference it at creation time. + let service_accounts = self.add_resources(service_accounts).await?; + let role_bindings = self.add_resources(role_bindings).await?; + let services = self.add_resources(services).await?; + let listeners = self.add_resources(listeners).await?; + let config_maps = self.add_resources(config_maps).await?; + let maybe_discovery_config_map = match maybe_discovery_config_map { + Some(config_map) => Some(self.add_resource(config_map).await?), + None => None, + }; + let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + let stateful_sets = self.add_resources(stateful_sets).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + maybe_discovery_config_map, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + applied_resources.push(self.add_resource(resource).await?); + } + + Ok(applied_resources) + } + + async fn add_resource(&mut self, resource: T) -> Result { + self.cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu) + } +} diff --git a/rust/operator-binary/src/zk_controller/build.rs b/rust/operator-binary/src/zk_controller/build.rs index f467fe40..e2af711b 100644 --- a/rust/operator-binary/src/zk_controller/build.rs +++ b/rust/operator-binary/src/zk_controller/build.rs @@ -9,7 +9,7 @@ //! remaining submodules ([`command`], [`graceful_shutdown`], [`jvm`], //! [`properties`]) produce fragments that those resource builders assemble. -use std::str::FromStr; +use std::{marker::PhantomData, str::FromStr}; use snafu::{ResultExt, Snafu}; use stackable_operator::{ @@ -20,8 +20,9 @@ use stackable_operator::{ use crate::{ crd::ZookeeperRole, + discovery, zk_controller::{ - KubernetesResources, + KubernetesResources, Prepared, ZK_CONTROLLER_NAME, build::resource::{ config_map, listener::build_role_listener, @@ -36,10 +37,6 @@ use crate::{ }, }; -// Placeholder role-group name used for the recommended labels of the role-level discovery -// `ConfigMap` (which is not tied to a single role group). -stackable_operator::constant!(pub(crate) PLACEHOLDER_DISCOVERY_ROLE_GROUP: RoleGroupName = "discovery"); - // Placeholder role-group name used for the recommended labels of the role-level `Listener` // (which is not tied to a single role group). stackable_operator::constant!(pub(crate) NONE_ROLE_GROUP_NAME: RoleGroupName = "none"); @@ -63,6 +60,9 @@ pub enum Error { source: statefulset::Error, rolegroup: RoleGroupName, }, + + #[snafu(display("failed to build the discovery ConfigMap"))] + DiscoveryConfigMap { source: discovery::Error }, } /// Builds every Kubernetes resource for the given validated cluster. @@ -72,13 +72,16 @@ pub enum Error { /// failures only. `cluster_info` is static cluster metadata (not a client call), consumed by the /// role-group ConfigMap builder. /// -/// The discovery `ConfigMap` is deliberately absent: it is built from the *applied* role -/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener)'s ingress addresses, so it -/// is assembled in the reconcile step after the Listener has been applied, not here. +/// The discovery `ConfigMap` is only built once the role +/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener) publishes ingress addresses. +/// Those are dereferenced and validated into +/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses) +/// before this step runs, so the ConfigMap is absent during the reconciliation that first creates +/// the Listener, and built by the one that the Listener watch triggers afterwards. pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, -) -> Result { +) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut config_maps = vec![]; @@ -127,14 +130,25 @@ pub fn build( let listeners = vec![build_role_listener(cluster, &zk_role)]; + let maybe_discovery_config_map = cluster + .discovery_addresses + .as_ref() + .map(|listener_addresses| { + discovery::build_discovery_configmap(cluster, ZK_CONTROLLER_NAME, listener_addresses) + }) + .transpose() + .context(DiscoveryConfigMapSnafu)?; + Ok(KubernetesResources { stateful_sets, services, listeners, config_maps, + maybe_discovery_config_map, pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } @@ -214,7 +228,7 @@ mod tests { "simple-zookeeper-server-secondary-metrics", ] ); - // One ConfigMap per role group; the discovery ConfigMap is absent — see `build()`. + // One ConfigMap per role group. assert_eq!( sorted_names(&resources.config_maps), [ @@ -222,6 +236,8 @@ mod tests { "simple-zookeeper-server-secondary", ] ); + // The fixture has no role Listener yet, so the discovery ConfigMap is absent (see `build()`). + assert!(resources.maybe_discovery_config_map.is_none()); // The single role-level Listener for the one ZooKeeper role (`server`). assert_eq!( sorted_names(&resources.listeners), diff --git a/rust/operator-binary/src/zk_controller/build/resource/discovery.rs b/rust/operator-binary/src/zk_controller/build/resource/discovery.rs deleted file mode 100644 index d1742478..00000000 --- a/rust/operator-binary/src/zk_controller/build/resource/discovery.rs +++ /dev/null @@ -1,288 +0,0 @@ -use std::{collections::BTreeSet, num::TryFromIntError, str::FromStr}; - -use snafu::{OptionExt, ResultExt, Snafu}; -use stackable_operator::{ - builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder}, - crd::listener, - k8s_openapi::api::core::v1::ConfigMap, - kube::{Resource, runtime::reflector::ObjectRef}, - v2::{ - HasName, HasUid, NameIsValidLabelValue, - builder::meta::ownerreference_from_resource, - kvp::label::recommended_labels, - types::operator::{ControllerName, ProductVersion}, - }, -}; - -use crate::{ - crd::{ZOOKEEPER_SERVER_PORT_NAME, ZookeeperRole, security::ZookeeperSecurity}, - zk_controller::{ - build::PLACEHOLDER_DISCOVERY_ROLE_GROUP, - validate::{ValidatedCluster, operator_name, product_name}, - }, - znode_controller::validate::ValidatedZnode, -}; - -type Result = std::result::Result; - -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display("chroot path {} was relative (must be absolute)", chroot))] - RelativeChroot { chroot: String }, - - #[snafu(display("{listener} does not have a port with the name {port_name:?}"))] - PortNotFound { - port_name: String, - listener: ObjectRef, - }, - - #[snafu(display("expected an unsigned 16-bit port, got {port_number}"))] - InvalidPort { - source: TryFromIntError, - port_number: i32, - }, - - #[snafu(display("{listener} has no ingress addresses"))] - NoListenerIngressAddresses { - listener: ObjectRef, - }, - - #[snafu(display("failed to build ConfigMap"))] - BuildConfigMap { - source: stackable_operator::builder::configmap::Error, - }, -} - -/// Build the discovery [`ConfigMap`] for the cluster controller from the -/// [`ValidatedCluster`]. -/// -/// The ConfigMap is owned by, and placed in the namespace of, the cluster. The image and security -/// settings are taken from the [`ValidatedCluster`] rather than being passed in separately. -pub fn build_discovery_configmap( - validated_cluster: &ValidatedCluster, - controller_name: &str, - listener: listener::v1alpha1::Listener, -) -> Result { - build_discovery_configmap_for_owner( - validated_cluster, - &validated_cluster.namespace, - controller_name, - &validated_cluster.product_version, - listener, - None, - &validated_cluster.cluster_config.zookeeper_security, - ) -} - -/// Build the discovery [`ConfigMap`] for the znode controller. -/// -/// The ConfigMap is owned by, and placed in the namespace of, the -/// [`ValidatedZnode`]. The product version and `zookeeper_security` originate from the referenced -/// cluster (via the validated znode), while `chroot` isolates the znode within the shared ZooKeeper -/// ensemble. -pub fn build_znode_discovery_configmap( - validated_znode: &ValidatedZnode, - controller_name: &str, - listener: listener::v1alpha1::Listener, - chroot: &str, -) -> Result { - build_discovery_configmap_for_owner( - validated_znode, - &validated_znode.namespace, - controller_name, - &validated_znode.product_version, - listener, - Some(chroot), - &validated_znode.zookeeper_security, - ) -} - -/// Build a discovery [`ConfigMap`] containing ZooKeeper connection details from a -/// [`listener::v1alpha1::Listener`]. -/// -/// `owner` owns the ConfigMap (the [`ZookeeperCluster`](crate::crd::v1alpha1::ZookeeperCluster) for the cluster -/// controller, or the [`ZookeeperZnode`](crate::crd::v1alpha1::ZookeeperZnode) for the znode controller) and -/// `namespace` is where the ConfigMap is placed. -fn build_discovery_configmap_for_owner( - owner: &(impl Resource + HasName + HasUid + NameIsValidLabelValue), - namespace: impl Into, - controller_name: &str, - product_version: &ProductVersion, - listener: listener::v1alpha1::Listener, - chroot: Option<&str>, - zookeeper_security: &ZookeeperSecurity, -) -> Result { - let name = owner.to_name(); - - // The discovery ConfigMap is a role-level resource of the `server` role, conventionally - // labelled with the `discovery` role group. The controller name differs between the cluster and - // znode controllers, so it is passed in and validated into the type-safe newtype here. - let controller_name = ControllerName::from_str(controller_name) - .expect("the controller name is a valid label value"); - let role_group_name = PLACEHOLDER_DISCOVERY_ROLE_GROUP.clone(); - - let listener_addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME)?; - - // Write a connection string of the format that Java ZooKeeper client expects: - // "{host1}:{port1},{host2:port2},.../{chroot}" - // See https://zookeeper.apache.org/doc/current/apidocs/zookeeper-server/org/apache/zookeeper/ZooKeeper.html#ZooKeeper-java.lang.String-int-org.apache.zookeeper.Watcher- - let listener_addresses = listener_addresses - .into_iter() - .map(|(host, port)| format!("{host}:{port}")) - .collect::>() - .join(","); - let mut conn_str = listener_addresses.clone(); - if let Some(chroot) = chroot { - if !chroot.starts_with('/') { - return RelativeChrootSnafu { chroot }.fail(); - } - conn_str.push_str(chroot); - } - ConfigMapBuilder::new() - .metadata( - ObjectMetaBuilder::new() - .name(name) - .namespace(namespace) - .ownerreference(ownerreference_from_resource(owner, None, Some(true))) - .with_labels(recommended_labels( - owner, - &product_name(), - product_version, - &operator_name(), - &controller_name, - &ZookeeperRole::Server.into(), - &role_group_name, - )) - .build(), - ) - .add_data("ZOOKEEPER", conn_str) - // Some clients don't support ZooKeeper's merged `hosts/chroot` format, so export them separately for these clients - .add_data("ZOOKEEPER_HOSTS", listener_addresses) - .add_data( - "ZOOKEEPER_CLIENT_PORT", - zookeeper_security.client_port().to_string(), - ) - .add_data("ZOOKEEPER_CHROOT", chroot.unwrap_or("/")) - .build() - .context(BuildConfigMapSnafu) -} - -/// Lists all listener address and port number pairs for a given `port_name` for Pods participating in the [`Listener`][1] -/// -/// This returns pairs of `(Address, Port)`, where address could be a hostname or IP address of a node, clusterIP or external -/// load balancer depending on the Service type. -/// -/// ## Errors -/// -/// An error will be returned if there is no address found for the `port_name`. -/// -/// [1]: listener::v1alpha1::Listener -// TODO (@NickLarsenNZ): Move this to stackable-operator, so it can be used as listener.addresses_for_port(port_name) -fn listener_addresses( - listener: &listener::v1alpha1::Listener, - port_name: &str, -) -> Result + use<>> { - // Get addresses port pairs for addresses that have a port with the name that matches the one we are interested in - let address_port_pairs = listener - .status - .as_ref() - .and_then(|listener_status| listener_status.ingress_addresses.as_ref()) - .context(NoListenerIngressAddressesSnafu { listener })? - .iter() - // Filter the addresses that have the port we are interested in (they likely all have it though) - .filter_map(|listener_ingress| { - Some(listener_ingress.address.clone()).zip(listener_ingress.ports.get(port_name)) - }) - // Convert the port from i32 to u16 - .map(|(listener_address, &port_number)| { - let port_number: u16 = port_number - .try_into() - .context(InvalidPortSnafu { port_number })?; - Ok((listener_address, port_number)) - }) - .collect::, _>>()?; - - // An empty list is considered an error - match address_port_pairs.is_empty() { - true => PortNotFoundSnafu { - port_name, - listener, - } - .fail(), - false => Ok(address_port_pairs), - } -} - -#[cfg(test)] -mod tests { - use std::collections::BTreeMap; - - use stackable_operator::{ - crd::listener::v1alpha1::{ - AddressType, Listener, ListenerIngress, ListenerSpec, ListenerStatus, - }, - k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, - }; - - use super::*; - - fn listener(ingress_addresses: Option>) -> Listener { - Listener { - metadata: ObjectMeta { - name: Some("test-listener".to_owned()), - ..ObjectMeta::default() - }, - spec: ListenerSpec::default(), - status: Some(ListenerStatus { - service_name: None, - ingress_addresses, - node_ports: None, - }), - } - } - - fn ingress(port: i32) -> ListenerIngress { - ListenerIngress { - address: "node-0".to_owned(), - address_type: AddressType::Hostname, - ports: BTreeMap::from([(ZOOKEEPER_SERVER_PORT_NAME.to_owned(), port)]), - } - } - - #[test] - fn listener_addresses_returns_host_port_pairs() { - let listener = listener(Some(vec![ingress(2181)])); - let pairs: Vec<_> = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) - .expect("addresses") - .into_iter() - .collect(); - assert_eq!(pairs, vec![("node-0".to_owned(), 2181u16)]); - } - - #[test] - fn listener_addresses_without_ingress_is_error() { - assert!(matches!( - listener_addresses(&listener(None), ZOOKEEPER_SERVER_PORT_NAME), - Err(Error::NoListenerIngressAddresses { .. }) - )); - } - - #[test] - fn listener_addresses_missing_port_name_is_error() { - let listener = listener(Some(vec![ingress(2181)])); - assert!(matches!( - listener_addresses(&listener, "does-not-exist"), - Err(Error::PortNotFound { .. }) - )); - } - - #[test] - fn listener_addresses_port_out_of_u16_range_is_error() { - // A port number that does not fit into a u16 must be rejected. - let listener = listener(Some(vec![ingress(70_000)])); - assert!(matches!( - listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME), - Err(Error::InvalidPort { .. }) - )); - } -} diff --git a/rust/operator-binary/src/zk_controller/build/resource/mod.rs b/rust/operator-binary/src/zk_controller/build/resource/mod.rs index 7f57f617..6b846018 100644 --- a/rust/operator-binary/src/zk_controller/build/resource/mod.rs +++ b/rust/operator-binary/src/zk_controller/build/resource/mod.rs @@ -2,7 +2,6 @@ //! into complete Kubernetes resources. pub mod config_map; -pub mod discovery; pub mod listener; pub mod pdb; pub mod rbac; diff --git a/rust/operator-binary/src/zk_controller/dereference.rs b/rust/operator-binary/src/zk_controller/dereference.rs index 38585d77..ce80dee7 100644 --- a/rust/operator-binary/src/zk_controller/dereference.rs +++ b/rust/operator-binary/src/zk_controller/dereference.rs @@ -6,17 +6,40 @@ //! validate step. use snafu::{ResultExt, Snafu}; -use stackable_operator::client::Client; +use stackable_operator::{ + client::Client, + crd::listener, + v2::{ + controller_utils::{get_cluster_name, get_namespace}, + types::{kubernetes::NamespaceName, operator::ClusterName}, + }, +}; use crate::crd::{ + ZookeeperRole, authentication::{self, DereferencedAuthenticationClasses}, - v1alpha1, + role_listener_name, v1alpha1, }; #[derive(Snafu, Debug)] pub enum Error { #[snafu(display("failed to fetch authentication classes"))] FetchAuthenticationClasses { source: authentication::Error }, + + #[snafu(display("failed to get the cluster name"))] + GetClusterName { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("failed to get the cluster namespace"))] + GetNamespace { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("failed to fetch the role Listener"))] + FetchRoleListener { + source: stackable_operator::client::Error, + }, } type Result = std::result::Result; @@ -25,6 +48,13 @@ type Result = std::result::Result; /// not yet validated. pub struct DereferencedObjects { pub authentication_classes: DereferencedAuthenticationClasses, + + /// The role Listener as created by an earlier reconciliation, if it exists already. + /// + /// The discovery ConfigMap advertises the addresses that the listener operator publishes on + /// this object, so it can only be built once the Listener exists and carries them. The + /// controller watches Listeners, so a reconciliation is triggered as soon as that happens. + pub maybe_role_listener: Option, } /// Fetches all Kubernetes objects referenced from the [`v1alpha1::ZookeeperCluster`] spec. @@ -32,6 +62,9 @@ pub async fn dereference( client: &Client, zk: &v1alpha1::ZookeeperCluster, ) -> Result { + let cluster_name = get_cluster_name(zk).context(GetClusterNameSnafu)?; + let namespace = get_namespace(zk).context(GetNamespaceSnafu)?; + let authentication_classes = DereferencedAuthenticationClasses::fetch_references( client, &zk.spec.cluster_config.authentication, @@ -39,7 +72,23 @@ pub async fn dereference( .await .context(FetchAuthenticationClassesSnafu)?; + let maybe_role_listener = fetch_role_listener(client, &cluster_name, &namespace).await?; + Ok(DereferencedObjects { authentication_classes, + maybe_role_listener, }) } + +async fn fetch_role_listener( + client: &Client, + cluster_name: &ClusterName, + namespace: &NamespaceName, +) -> Result> { + let listener_name = role_listener_name(cluster_name.as_ref(), &ZookeeperRole::Server); + + client + .get_opt(listener_name.as_ref(), namespace.as_ref()) + .await + .context(FetchRoleListenerSnafu) +} diff --git a/rust/operator-binary/src/zk_controller/update_status.rs b/rust/operator-binary/src/zk_controller/update_status.rs new file mode 100644 index 00000000..f45e1b21 --- /dev/null +++ b/rust/operator-binary/src/zk_controller/update_status.rs @@ -0,0 +1,89 @@ +//! The update_status step in the ZookeeperCluster controller. + +use std::hash::Hasher; + +use fnv::FnvHasher; +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + OPERATOR_NAME, + crd::v1alpha1, + zk_controller::{Applied, KubernetesResources}, +}; + +#[derive(Snafu, Debug, EnumDiscriminants)] +#[strum_discriminants(derive(IntoStaticStr))] +pub enum Error { + #[snafu(display("failed to update status"))] + ApplyStatus { + source: stackable_operator::client::Error, + }, +} + +type Result = std::result::Result; + +/// Computes the cluster status from the applied resources and patches it onto the +/// [`v1alpha1::ZookeeperCluster`]. +/// +/// Takes [`KubernetesResources`] so the type system proves that the status derives from +/// applied resources, not merely built ones. +pub async fn update_status( + client: &Client, + zk: &v1alpha1::ZookeeperCluster, + applied: &KubernetesResources, +) -> Result<()> { + let mut stateful_set_condition_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + stateful_set_condition_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&zk.spec.cluster_operation); + + let status = v1alpha1::ZookeeperClusterStatus { + discovery_hash: Some(discovery_hash(applied)), + conditions: compute_conditions( + zk, + &[ + &stateful_set_condition_builder, + &cluster_operation_cond_builder, + ], + ), + }; + + client + .apply_patch_status(OPERATOR_NAME, zk, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} + +/// Hashes the resource version of the applied discovery ConfigMap, so that clients can tell when +/// the published connection details changed. +/// +/// The hash covers nothing while the discovery ConfigMap is absent, which is the case until the +/// role Listener publishes its addresses. +fn discovery_hash(applied: &KubernetesResources) -> String { + // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. + // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. + let mut discovery_hash = FnvHasher::with_key(0); + + if let Some(discovery_config_map) = &applied.maybe_discovery_config_map + && let Some(resource_version) = &discovery_config_map.metadata.resource_version + { + discovery_hash.write(resource_version.as_bytes()) + } + + // Serialize as a string to discourage users from trying to parse the value, + // and to keep things flexible if we end up changing the hasher at some point. + discovery_hash.finish().to_string() +} diff --git a/rust/operator-binary/src/zk_controller/validate.rs b/rust/operator-binary/src/zk_controller/validate.rs index 93300a46..506dd683 100644 --- a/rust/operator-binary/src/zk_controller/validate.rs +++ b/rust/operator-binary/src/zk_controller/validate.rs @@ -51,11 +51,12 @@ use strum::IntoEnumIterator; use crate::{ crd::{ - APP_NAME, CONTAINER_IMAGE_BASE_NAME, OPERATOR_NAME, ZookeeperRole, ZookeeperServerRoleType, - authentication, + APP_NAME, CONTAINER_IMAGE_BASE_NAME, OPERATOR_NAME, ZOOKEEPER_SERVER_PORT_NAME, + ZookeeperRole, ZookeeperServerRoleType, authentication, security::ZookeeperSecurity, v1alpha1::{self, ZookeeperConfig, ZookeeperConfigOverrides, ZookeeperServerRoleConfig}, }, + listener_addresses::{self, ListenerAddresses, listener_addresses}, zk_controller::{ZK_CONTROLLER_NAME, dereference::DereferencedObjects}, }; @@ -117,6 +118,9 @@ pub enum Error { "the Vector agent is enabled but no Vector aggregator discovery ConfigMap name is set" ))] MissingVectorAggregatorConfigMapName, + + #[snafu(display("failed to read the addresses published by the role Listener"))] + ReadRoleListenerAddresses { source: listener_addresses::Error }, } type Result = std::result::Result; @@ -242,6 +246,12 @@ pub struct ValidatedCluster { /// Object overrides applied to the cluster's resources, carried so the apply step does not reach /// into the raw [`v1alpha1::ZookeeperCluster`]. pub object_overrides: ObjectOverrides, + /// The client addresses published by the role Listener, which the discovery ConfigMap + /// advertises. + /// + /// `None` until the listener operator has published them, in which case the discovery + /// ConfigMap is skipped and built by the reconciliation that the Listener watch triggers. + pub discovery_addresses: Option, } // Placeholder product version used for labels on PVC templates, which cannot be modified once @@ -264,6 +274,7 @@ impl ValidatedCluster { >, cluster_operation: ClusterOperation, object_overrides: ObjectOverrides, + discovery_addresses: Option, ) -> Self { Self { metadata: ObjectMeta { @@ -282,6 +293,7 @@ impl ValidatedCluster { role_group_configs, cluster_operation, object_overrides, + discovery_addresses, } } @@ -508,6 +520,16 @@ pub fn validate( pdb: common.pod_disruption_budget.clone(), }; + // The role Listener does not exist during the very first reconciliation, and carries no + // addresses until the listener operator has published them. + let discovery_addresses = dereferenced_objects + .maybe_role_listener + .as_ref() + .map(|listener| listener_addresses(listener, ZOOKEEPER_SERVER_PORT_NAME)) + .transpose() + .context(ReadRoleListenerAddressesSnafu)? + .flatten(); + Ok(ValidatedCluster::new( name, namespace, @@ -522,6 +544,7 @@ pub fn validate( role_group_configs, zk.spec.cluster_operation.clone(), zk.spec.object_overrides.clone(), + discovery_addresses, )) } diff --git a/rust/operator-binary/src/znode_controller.rs b/rust/operator-binary/src/znode_controller.rs index c0905988..831487c5 100644 --- a/rust/operator-binary/src/znode_controller.rs +++ b/rust/operator-binary/src/znode_controller.rs @@ -1,17 +1,20 @@ //! Reconciles state for ZooKeeper znodes between Kubernetes [`v1alpha1::ZookeeperZnode`] objects and the ZooKeeper cluster //! //! See [`v1alpha1::ZookeeperZnode`] for more details. +//! +//! This is the controller driver: it runs the `dereference -> validate -> build -> apply` +//! pipeline, with each step living in its own submodule. There is no update_status step, because +//! the only status the ZookeeperZnode carries (the znode path) is written before the finalizer +//! runs. use std::{borrow::Cow, convert::Infallible, sync::Arc}; use const_format::concatcp; use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, - cluster_resources::{ClusterResourceApplyStrategy, ClusterResources}, - crd::listener, + cluster_resources::ClusterResourceApplyStrategy, k8s_openapi::api::core::v1::ConfigMap, kube::{ - Resource, ResourceExt, api::ObjectMeta, core::{DeserializeGuard, DynamicObject, error_boundary}, runtime::{controller, finalizer, reflector::ObjectRef}, @@ -24,11 +27,13 @@ use strum::{EnumDiscriminants, IntoStaticStr}; use tracing::{debug, info}; use crate::{ - APP_NAME, OPERATOR_NAME, - crd::{ZookeeperRole, role_listener_name, security::ZookeeperSecurity, v1alpha1}, - zk_controller::build::resource::discovery::{self, build_znode_discovery_configmap}, + OPERATOR_NAME, + crd::{security::ZookeeperSecurity, v1alpha1}, + znode_controller::apply::{Applier, ensure_znode_exists}, }; +pub(crate) mod apply; +pub(crate) mod build; mod dereference; pub(crate) mod validate; @@ -60,24 +65,11 @@ pub enum Error { ))] ObjectMissingMetadata, - #[snafu(display("could not find server role service for {zk:?}"))] - FindZkSvc { - source: stackable_operator::client::Error, - zk: ObjectRef, - }, - #[snafu(display("failed to calculate FQDN for {zk:?}"))] NoZkFqdn { zk: ObjectRef, }, - #[snafu(display("failed to ensure that ZNode {znode_path:?} exists in {zk:?}"))] - EnsureZnode { - source: znode_mgmt::Error, - zk: ObjectRef, - znode_path: String, - }, - #[snafu(display("failed to ensure that ZNode {znode_path:?} is missing from {zk:?}"))] EnsureZnodeMissing { source: znode_mgmt::Error, @@ -85,14 +77,11 @@ pub enum Error { znode_path: String, }, - #[snafu(display("failed to build discovery information"))] - BuildDiscoveryConfigMap { source: discovery::Error }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to save discovery information to {cm:?}"))] - ApplyDiscoveryConfigMap { - source: stackable_operator::cluster_resources::Error, - cm: ObjectRef, - }, + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, #[snafu(display("failed to update status"))] ApplyStatus { @@ -103,14 +92,6 @@ pub enum Error { Finalizer { source: finalizer::Error, }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphans { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("object has no namespace"))] - ObjectHasNoNamespace, } type Result = std::result::Result; @@ -146,20 +127,24 @@ impl ReconcilerError for Error { Error::Dereference { .. } => None, Error::ValidateCluster { .. } => None, Error::ObjectMissingMetadata => None, - Error::FindZkSvc { zk, .. } => Some(zk.clone().erase()), Error::NoZkFqdn { zk } => Some(zk.clone().erase()), - Error::EnsureZnode { zk, .. } => Some(zk.clone().erase()), Error::EnsureZnodeMissing { zk, .. } => Some(zk.clone().erase()), - Error::BuildDiscoveryConfigMap { .. } => None, - Error::ApplyDiscoveryConfigMap { cm, .. } => Some(cm.clone().erase()), + Error::BuildResources { .. } => None, + Error::ApplyResources { .. } => None, Error::ApplyStatus { .. } => None, Error::Finalizer { .. } => None, - Error::DeleteOrphans { .. } => None, - Error::ObjectHasNoNamespace => None, } } } +/// Every Kubernetes resource produced by the client-free [`build()`](build::build) step. +/// +/// The znode path inside the ZooKeeper ensemble is not a Kubernetes object, so it is absent here +/// and created by the apply step instead. +pub struct KubernetesResources { + pub discovery_config_maps: Vec, +} + pub async fn reconcile_znode( znode: Arc>, ctx: Arc, @@ -260,23 +245,8 @@ async fn reconcile_apply( zk: v1alpha1::ZookeeperCluster, znode_path: &str, ) -> Result { - // Infallible: `ValidatedZnode`'s object reference always contains name, namespace and uid - // (set unconditionally during the validate step), which is all `ClusterResources::new` - // requires. - let mut cluster_resources = ClusterResources::new( - APP_NAME, - OPERATOR_NAME, - ZNODE_CONTROLLER_NAME, - &validated_znode.object_ref(&()), - ClusterResourceApplyStrategy::from(&validated_znode.cluster_operation), - &validated_znode.object_overrides, - ) - .expect( - "ClusterResources should be created because the ValidatedZnode's object reference \ - always contains name, namespace and uid", - ); - - znode_mgmt::ensure_znode_exists( + // The znode must exist in the ZooKeeper ensemble before the discovery ConfigMap advertises it. + ensure_znode_exists( &zk_mgmt_addr( &zk, &validated_znode.zookeeper_security, @@ -285,42 +255,22 @@ async fn reconcile_apply( znode_path, ) .await - .with_context(|_| EnsureZnodeSnafu { - zk: ObjectRef::from_obj(&zk), - znode_path, - })?; + .context(ApplyResourcesSnafu)?; + + // build (no client required) + let resources = build::build(validated_znode, znode_path).context(BuildResourcesSnafu)?; - let listener = client - .get::( - role_listener_name(&zk.name_any(), &ZookeeperRole::Server).as_ref(), - zk.metadata - .namespace - .as_deref() - .context(ObjectHasNoNamespaceSnafu)?, - ) - .await - .context(FindZkSvcSnafu { - zk: ObjectRef::from_obj(&zk), - })?; - - let discovery_cm = build_znode_discovery_configmap( + // apply (client required) + Applier::new( + client, validated_znode, - ZNODE_CONTROLLER_NAME, - listener, - znode_path, + ClusterResourceApplyStrategy::from(&validated_znode.cluster_operation), + &validated_znode.object_overrides, ) - .context(BuildDiscoveryConfigMapSnafu)?; - - let obj_ref = ObjectRef::from_obj(&discovery_cm); - cluster_resources - .add(client, discovery_cm) - .await - .with_context(|_| ApplyDiscoveryConfigMapSnafu { cm: obj_ref })?; - - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphansSnafu)?; + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; + Ok(controller::Action::await_change()) } diff --git a/rust/operator-binary/src/znode_controller/apply.rs b/rust/operator-binary/src/znode_controller/apply.rs new file mode 100644 index 00000000..c96106e0 --- /dev/null +++ b/rust/operator-binary/src/znode_controller/apply.rs @@ -0,0 +1,131 @@ +//! The apply step in the ZookeeperZnode controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + kube::Resource, +}; + +use crate::{ + APP_NAME, OPERATOR_NAME, + znode_controller::{ + KubernetesResources, ZNODE_CONTROLLER_NAME, validate::ValidatedZnode, znode_mgmt, + }, +}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { + source: stackable_operator::cluster_resources::Error, + }, + + #[snafu(display("failed to delete orphaned resources"))] + DeleteOrphanedResources { + source: stackable_operator::cluster_resources::Error, + }, + + #[snafu(display("failed to ensure that ZNode {znode_path:?} exists in {zk_mgmt_addr}"))] + EnsureZnode { + source: znode_mgmt::Error, + zk_mgmt_addr: String, + znode_path: String, + }, +} + +type Result = std::result::Result; + +/// Applier for the Kubernetes resource specifications produced by this controller. +/// +/// Unlike the ZookeeperCluster controller's applier, the resources are not marked as prepared or +/// applied: the ZookeeperZnode has no cluster conditions, so there is no status step that could +/// derive the status from resources that were never applied. +pub struct Applier<'a> { + client: &'a Client, + cluster_resources: ClusterResources<'a>, +} + +impl<'a> Applier<'a> { + pub fn new( + client: &'a Client, + znode: &ValidatedZnode, + apply_strategy: ClusterResourceApplyStrategy, + object_overrides: &'a ObjectOverrides, + ) -> Applier<'a> { + // Infallible: `ValidatedZnode`'s object reference always contains name, namespace and uid + // (set unconditionally during the validate step), which is all `ClusterResources::new` + // requires. + let cluster_resources = ClusterResources::new( + APP_NAME, + OPERATOR_NAME, + ZNODE_CONTROLLER_NAME, + &znode.object_ref(&()), + apply_strategy, + object_overrides, + ) + .expect( + "ClusterResources should be created because the ValidatedZnode's object reference \ + always contains name, namespace and uid", + ); + + Applier { + client, + cluster_resources, + } + } + + /// Applies the given Kubernetes resources. + pub async fn apply(mut self, resources: KubernetesResources) -> Result { + // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to + // compile here instead of silently never being applied. + let KubernetesResources { + discovery_config_maps, + } = resources; + + let discovery_config_maps = self.add_resources(discovery_config_maps).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + discovery_config_maps, + }) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + applied_resources.push( + self.cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu)?, + ); + } + + Ok(applied_resources) + } +} + +/// Ensures that the znode exists in the ZooKeeper ensemble reachable at `zk_mgmt_addr`. +/// +/// The znode is a path inside ZooKeeper rather than a Kubernetes object, so it cannot be part of +/// the client-free `build()` step, and it is not tracked in +/// [`ClusterResources`](stackable_operator::cluster_resources::ClusterResources) either. It must +/// exist before the discovery ConfigMap advertises it to clients. +pub async fn ensure_znode_exists(zk_mgmt_addr: &str, znode_path: &str) -> Result<()> { + znode_mgmt::ensure_znode_exists(zk_mgmt_addr, znode_path) + .await + .with_context(|_| EnsureZnodeSnafu { + zk_mgmt_addr, + znode_path, + }) +} diff --git a/rust/operator-binary/src/znode_controller/build.rs b/rust/operator-binary/src/znode_controller/build.rs new file mode 100644 index 00000000..509b4adb --- /dev/null +++ b/rust/operator-binary/src/znode_controller/build.rs @@ -0,0 +1,33 @@ +//! The build step in the ZookeeperZnode controller. + +use snafu::{ResultExt, Snafu}; + +use crate::{ + discovery::{self, build_znode_discovery_configmap}, + znode_controller::{KubernetesResources, ZNODE_CONTROLLER_NAME, validate::ValidatedZnode}, +}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build the discovery ConfigMap"))] + DiscoveryConfigMap { source: discovery::Error }, +} + +/// Builds every Kubernetes resource for the given validated znode. +/// +/// Does not need a Kubernetes client: the referenced cluster and the addresses published by its +/// role Listener are already dereferenced and validated by this point. The znode itself (a path +/// inside the ZooKeeper ensemble, not a Kubernetes object) is created by the apply step. +pub fn build(znode: &ValidatedZnode, znode_path: &str) -> Result { + let discovery_config_map = build_znode_discovery_configmap( + znode, + ZNODE_CONTROLLER_NAME, + &znode.discovery_addresses, + znode_path, + ) + .context(DiscoveryConfigMapSnafu)?; + + Ok(KubernetesResources { + discovery_config_maps: vec![discovery_config_map], + }) +} diff --git a/rust/operator-binary/src/znode_controller/dereference.rs b/rust/operator-binary/src/znode_controller/dereference.rs index d29ee402..1d0a4055 100644 --- a/rust/operator-binary/src/znode_controller/dereference.rs +++ b/rust/operator-binary/src/znode_controller/dereference.rs @@ -1,19 +1,21 @@ //! The dereference step in the ZookeeperZnode controller. //! //! Fetches the parent [`v1alpha1::ZookeeperCluster`] referenced by the znode's -//! `spec.clusterRef`, plus the [`DereferencedAuthenticationClasses`] of that cluster. Both Apply -//! and Cleanup paths in `reconcile_znode` share this output. Synchronous validation of the -//! fetched objects happens in the validate step. +//! `spec.clusterRef`, plus the [`DereferencedAuthenticationClasses`] and the role Listener of that +//! cluster. Both Apply and Cleanup paths in `reconcile_znode` share this output. Synchronous +//! validation of the fetched objects happens in the validate step. -use snafu::{ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ client::Client, - kube::{self, runtime::reflector::ObjectRef}, + crd::listener, + kube::{self, ResourceExt, runtime::reflector::ObjectRef}, }; use crate::crd::{ + ZookeeperRole, authentication::{self, DereferencedAuthenticationClasses}, - v1alpha1, + role_listener_name, v1alpha1, }; #[derive(Snafu, Debug)] @@ -35,6 +37,17 @@ pub enum Error { #[snafu(display("failed to fetch authentication classes"))] FetchAuthenticationClasses { source: authentication::Error }, + + #[snafu(display("{zk} has no namespace"))] + ZkHasNoNamespace { + zk: ObjectRef, + }, + + #[snafu(display("failed to fetch the role Listener of {zk}"))] + FetchRoleListener { + source: stackable_operator::client::Error, + zk: ObjectRef, + }, } type Result = std::result::Result; @@ -43,6 +56,13 @@ type Result = std::result::Result; pub struct DereferencedObjects { pub zk: v1alpha1::ZookeeperCluster, pub authentication_classes: DereferencedAuthenticationClasses, + + /// The role Listener of the referenced cluster, if it exists already. + /// + /// The znode's discovery ConfigMap advertises the addresses that the listener operator + /// publishes on it. The Cleanup path does not need it, so a missing Listener is not an error + /// here and never blocks finalizer removal. + pub maybe_role_listener: Option, } /// Fetches all Kubernetes objects referenced from the [`v1alpha1::ZookeeperZnode`] spec. @@ -59,12 +79,33 @@ pub async fn dereference( .await .context(FetchAuthenticationClassesSnafu)?; + let maybe_role_listener = fetch_role_listener(client, &zk).await?; + Ok(DereferencedObjects { zk, authentication_classes, + maybe_role_listener, }) } +async fn fetch_role_listener( + client: &Client, + zk: &v1alpha1::ZookeeperCluster, +) -> Result> { + let zk_ref = ObjectRef::from_obj(zk); + let namespace = zk + .metadata + .namespace + .as_deref() + .with_context(|| ZkHasNoNamespaceSnafu { zk: zk_ref.clone() })?; + let listener_name = role_listener_name(&zk.name_any(), &ZookeeperRole::Server); + + client + .get_opt(listener_name.as_ref(), namespace) + .await + .with_context(|_| FetchRoleListenerSnafu { zk: zk_ref }) +} + async fn find_zk_of_znode( client: &Client, znode: &v1alpha1::ZookeeperZnode, diff --git a/rust/operator-binary/src/znode_controller/validate.rs b/rust/operator-binary/src/znode_controller/validate.rs index 899b6cdb..570721c2 100644 --- a/rust/operator-binary/src/znode_controller/validate.rs +++ b/rust/operator-binary/src/znode_controller/validate.rs @@ -24,7 +24,11 @@ use stackable_operator::{ }; use crate::{ - crd::{CONTAINER_IMAGE_BASE_NAME, authentication, security::ZookeeperSecurity, v1alpha1}, + crd::{ + CONTAINER_IMAGE_BASE_NAME, ZOOKEEPER_SERVER_PORT_NAME, authentication, + security::ZookeeperSecurity, v1alpha1, + }, + listener_addresses::{self, ListenerAddresses, listener_addresses}, znode_controller::dereference::DereferencedObjects, }; @@ -62,6 +66,14 @@ pub enum Error { source: stackable_operator::v2::macros::attributed_string_type::Error, product_version: String, }, + + #[snafu(display("failed to read the addresses published by the ZooKeeper role Listener"))] + ReadRoleListenerAddresses { source: listener_addresses::Error }, + + #[snafu(display( + "the ZooKeeper role Listener does not exist yet, or has not published any addresses yet" + ))] + NoRoleListenerAddresses, } type Result = std::result::Result; @@ -92,6 +104,12 @@ pub struct ValidatedZnode { /// Object overrides applied to the znode's resources, carried so the apply step does not reach /// into the raw [`v1alpha1::ZookeeperZnode`]. pub object_overrides: ObjectOverrides, + /// The client addresses published by the referenced cluster's role Listener, which the znode's + /// discovery ConfigMap advertises. + /// + /// Unlike the cluster controller, the znode controller cannot produce anything without them, + /// so validation fails while they are missing and the reconciliation is retried. + pub discovery_addresses: ListenerAddresses, } impl HasName for ValidatedZnode { @@ -184,6 +202,15 @@ pub fn validate( } })?; + let discovery_addresses = dereferenced_objects + .maybe_role_listener + .as_ref() + .map(|listener| listener_addresses(listener, ZOOKEEPER_SERVER_PORT_NAME)) + .transpose() + .context(ReadRoleListenerAddressesSnafu)? + .flatten() + .context(NoRoleListenerAddressesSnafu)?; + Ok(ValidatedZnode { metadata: ObjectMeta { name: Some(name.clone()), @@ -198,5 +225,6 @@ pub fn validate( zookeeper_security, cluster_operation: dereferenced_objects.zk.spec.cluster_operation.clone(), object_overrides: znode.spec.object_overrides.clone(), + discovery_addresses, }) }