diff --git a/CHANGELOG.md b/CHANGELOG.md index 380c71f1..4683c192 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,10 @@ 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]). +- The discovery ConfigMap is now always written, with empty `ZOOKEEPER` and `ZOOKEEPER_HOSTS` while + the listener publishes no addresses ([#1069]). - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#1070]). ### Fixed @@ -29,6 +33,7 @@ All notable changes to this project will be documented in this file. [#1060]: https://github.com/stackabletech/zookeeper-operator/pull/1060 [#1063]: https://github.com/stackabletech/zookeeper-operator/pull/1063 [#1068]: https://github.com/stackabletech/zookeeper-operator/pull/1068 +[#1069]: https://github.com/stackabletech/zookeeper-operator/pull/1069 [#1070]: https://github.com/stackabletech/zookeeper-operator/pull/1070 ## [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..9b3a7d8b --- /dev/null +++ b/rust/operator-binary/src/discovery.rs @@ -0,0 +1,325 @@ +//! 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. +/// +/// The connection details are read from the addresses published by the role Listener (carried on +/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses), +/// fetched in the dereference step), which only the listener operator writes. While no address +/// exists around the first reconciliations, which create the Listener in the first place, the +/// ConfigMap is still written, with an empty `ZOOKEEPER` value: omitting it instead would let the +/// apply step delete an existing discovery ConfigMap as an orphan, breaking consumers that mount +/// it. The Listener watch triggers a new run that fills in the value once the addresses are +/// published. +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 the +/// [`ListenerAddresses`] published by the role 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) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + crd::ZOOKEEPER_SERVER_PORT_NAME, + listener_addresses::{ + listener_addresses, + test_support::{ingress_address, role_listener}, + }, + zk_controller::{ + ZK_CONTROLLER_NAME, + test_support::{minimal_zk, try_validate_with_role_listener}, + }, + znode_controller::{ + ZNODE_CONTROLLER_NAME, + test_support::{minimal_znode, validated_znode}, + }, + }; + + const ZK_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + "#; + + const ZNODE_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperZnode + metadata: + name: simple-znode + spec: + clusterRef: + name: simple-zookeeper + "#; + + /// The znode path that the znode controller derives from the fixture's UID. + const ZNODE_PATH: &str = "/znode-e5dbf9c2-d8b0-4c1e-9f4a-1d2e3f4a5b6c"; + + /// The value of `key` in the given discovery ConfigMap. + fn data(config_map: &ConfigMap, key: &str) -> String { + config_map + .data + .as_ref() + .expect("the discovery ConfigMap should carry data") + .get(key) + .unwrap_or_else(|| panic!("the discovery ConfigMap should carry {key}")) + .clone() + } + + /// Addresses published under the ZooKeeper server port name, built through the real reader so + /// the fixtures cannot drift from what the validate step produces. + fn published_addresses(addresses: &[(&str, u16)]) -> ListenerAddresses { + let listener = role_listener(Some( + addresses + .iter() + .map(|(address, port)| { + ingress_address(address, ZOOKEEPER_SERVER_PORT_NAME, i32::from(*port)) + }) + .collect(), + )); + + listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) + .expect("the fixture publishes addresses under the server port name") + .expect("the fixture publishes addresses") + } + + /// The cluster controller advertises the whole ensemble, rooted at `/`. The fixture keeps TLS + /// enabled, so the client port is the secure one. + #[test] + fn cluster_discovery_config_map_advertises_the_published_addresses() { + let cluster = try_validate_with_role_listener(&minimal_zk(ZK_YAML), None) + .expect("validate should succeed for the test fixture"); + let addresses = published_addresses(&[("node-0", 2282), ("node-1", 2282)]); + + let config_map = + build_discovery_configmap(&cluster, ZK_CONTROLLER_NAME, &addresses).expect("build"); + + assert_eq!( + config_map.metadata.name.as_deref(), + Some("simple-zookeeper") + ); + assert_eq!(config_map.metadata.namespace.as_deref(), Some("default")); + assert_eq!(data(&config_map, "ZOOKEEPER"), "node-0:2282,node-1:2282"); + assert_eq!( + data(&config_map, "ZOOKEEPER_HOSTS"), + "node-0:2282,node-1:2282" + ); + assert_eq!(data(&config_map, "ZOOKEEPER_CLIENT_PORT"), "2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CHROOT"), "/"); + } + + /// While the role Listener publishes no addresses the ConfigMap is still written, with an + /// empty connection string, so that the apply step does not delete the published one as an + /// orphan. + #[test] + fn cluster_discovery_config_map_without_addresses_is_still_written() { + let cluster = try_validate_with_role_listener(&minimal_zk(ZK_YAML), None) + .expect("validate should succeed for the test fixture"); + + let config_map = + build_discovery_configmap(&cluster, ZK_CONTROLLER_NAME, &ListenerAddresses::default()) + .expect("build"); + + assert_eq!( + config_map.metadata.name.as_deref(), + Some("simple-zookeeper") + ); + assert_eq!(data(&config_map, "ZOOKEEPER"), ""); + assert_eq!(data(&config_map, "ZOOKEEPER_HOSTS"), ""); + // The port and chroot do not depend on the addresses, so they stay populated. + assert_eq!(data(&config_map, "ZOOKEEPER_CLIENT_PORT"), "2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CHROOT"), "/"); + } + + /// The znode controller advertises the same ensemble, narrowed to the znode's chroot. Only + /// `ZOOKEEPER` carries the chroot, because some clients cannot parse the merged format. + #[test] + fn znode_discovery_config_map_narrows_the_ensemble_to_the_chroot() { + let znode = validated_znode(&minimal_znode(ZNODE_YAML)); + + let config_map = build_znode_discovery_configmap( + &znode, + ZNODE_CONTROLLER_NAME, + &znode.discovery_addresses, + ZNODE_PATH, + ) + .expect("build"); + + // The ConfigMap is named after the znode, not after the referenced cluster. + assert_eq!(config_map.metadata.name.as_deref(), Some("simple-znode")); + assert_eq!(config_map.metadata.namespace.as_deref(), Some("default")); + assert_eq!( + data(&config_map, "ZOOKEEPER"), + format!("node-0:2282{ZNODE_PATH}") + ); + assert_eq!(data(&config_map, "ZOOKEEPER_HOSTS"), "node-0:2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CLIENT_PORT"), "2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CHROOT"), ZNODE_PATH); + } + + /// A relative chroot would silently produce a connection string pointing at the ensemble root. + #[test] + fn relative_chroot_is_rejected() { + let znode = validated_znode(&minimal_znode(ZNODE_YAML)); + + assert!(matches!( + build_znode_discovery_configmap( + &znode, + ZNODE_CONTROLLER_NAME, + &znode.discovery_addresses, + "znode-without-a-leading-slash", + ), + Err(Error::RelativeChroot { .. }) + )); + } +} diff --git a/rust/operator-binary/src/listener_addresses.rs b/rust/operator-binary/src/listener_addresses.rs new file mode 100644 index 00000000..fff6c18b --- /dev/null +++ b/rust/operator-binary/src/listener_addresses.rs @@ -0,0 +1,184 @@ +//! 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. +/// +/// The [`Default`] value is empty, which renders as an empty connection string, see +/// [`build_discovery_configmap`](crate::discovery::build_discovery_configmap). +#[derive(Clone, Debug, Default, 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))), + } +} + +/// Shared helpers for building role [`Listener`](listener::v1alpha1::Listener) fixtures. +#[cfg(test)] +pub(crate) mod test_support { + use std::collections::BTreeMap; + + use stackable_operator::{ + crd::listener::v1alpha1::{ + AddressType, Listener, ListenerIngress, ListenerSpec, ListenerStatus, + }, + k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + + /// A role Listener publishing `ingress_addresses`. `None` is a Listener that the listener + /// operator has not written a status for yet. + pub fn role_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, + }), + } + } + + /// A single ingress address of a role Listener, publishing `port` under `port_name`. + pub fn ingress_address(address: &str, port_name: &str, port: i32) -> ListenerIngress { + ListenerIngress { + address: address.to_owned(), + address_type: AddressType::Hostname, + ports: BTreeMap::from([(port_name.to_owned(), port)]), + } + } +} + +#[cfg(test)] +mod tests { + use stackable_operator::crd::listener::v1alpha1::ListenerIngress; + + use super::{ + test_support::{ingress_address, role_listener}, + *, + }; + use crate::crd::ZOOKEEPER_SERVER_PORT_NAME; + + fn ingress(port: i32) -> ListenerIngress { + ingress_address("node-0", ZOOKEEPER_SERVER_PORT_NAME, port) + } + + #[test] + fn listener_addresses_returns_host_port_pairs() { + let listener = role_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(&role_listener(None), ZOOKEEPER_SERVER_PORT_NAME) + .expect("addresses"), + None + ); + } + + #[test] + fn listener_addresses_missing_port_name_is_error() { + let listener = role_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 = role_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..f246b8d5 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 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 apply the Kubernetes resources"))] + ApplyResources { source: apply::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,35 @@ 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`, kept apart from the role group `config_maps` because the cluster + /// status carries a hash of it. + pub discovery_config_map: ConfigMap, pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } pub async fn reconcile_zk( @@ -167,116 +138,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()) } @@ -298,7 +178,7 @@ pub fn error_policy( #[cfg(test)] pub(crate) mod test_support { use stackable_operator::{ - cli::OperatorEnvironmentOptions, commons::networking::DomainName, + cli::OperatorEnvironmentOptions, commons::networking::DomainName, crd::listener, utils::cluster_info::KubernetesClusterInfo, }; @@ -344,7 +224,7 @@ pub(crate) mod test_support { } } - fn operator_environment() -> OperatorEnvironmentOptions { + pub fn operator_environment() -> OperatorEnvironmentOptions { OperatorEnvironmentOptions { operator_namespace: "stackable-operators".to_owned(), operator_service_name: "zookeeper-operator".to_owned(), @@ -352,15 +232,25 @@ pub(crate) mod test_support { } } - /// Runs the real validate step against a minimal (auth-free) fixture, returning the result so - /// tests can assert on validation errors. + /// Runs the real validate step against a minimal (auth-free) fixture whose role Listener does + /// not exist yet, returning the result so tests can assert on validation errors. pub fn try_validate( zk: &v1alpha1::ZookeeperCluster, + ) -> Result { + try_validate_with_role_listener(zk, None) + } + + /// Runs the real validate step against a minimal (auth-free) fixture and the given role + /// Listener, returning the result so tests can assert on validation errors. + pub fn try_validate_with_role_listener( + zk: &v1alpha1::ZookeeperCluster, + maybe_role_listener: Option, ) -> Result { validate( zk, &DereferencedObjects { authentication_classes: DereferencedAuthenticationClasses::new_for_tests(), + maybe_role_listener, }, &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..9169e09a --- /dev/null +++ b/rust/operator-binary/src/zk_controller/apply.rs @@ -0,0 +1,138 @@ +//! 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, + 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 discovery_config_map = self.add_resource(discovery_config_map).await?; + 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, + 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..b762fc4d 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. +/// This includes the discovery `ConfigMap`, built from the addresses published by the role +/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener) that were dereferenced and +/// validated into +/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses); see +/// [`build_discovery_configmap`](discovery::build_discovery_configmap) for how its content depends +/// on them. 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,23 @@ pub fn build( let listeners = vec![build_role_listener(cluster, &zk_role)]; + let discovery_config_map = discovery::build_discovery_configmap( + cluster, + ZK_CONTROLLER_NAME, + &cluster.discovery_addresses, + ) + .context(DiscoveryConfigMapSnafu)?; + Ok(KubernetesResources { stateful_sets, services, listeners, config_maps, + discovery_config_map, pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } @@ -214,7 +226,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 +234,12 @@ mod tests { "simple-zookeeper-server-secondary", ] ); + // The discovery ConfigMap is named after the cluster, and written even though the fixture + // has no role Listener publishing addresses yet (see `build()`). + assert_eq!( + resources.discovery_config_map.meta().name.as_deref(), + Some("simple-zookeeper") + ); // 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..455d34a8 --- /dev/null +++ b/rust/operator-binary/src/zk_controller/update_status.rs @@ -0,0 +1,120 @@ +//! The update_status step in the ZookeeperCluster controller. + +use std::hash::Hasher; + +use fnv::FnvHasher; +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + k8s_openapi::api::core::v1::ConfigMap, + 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.discovery_config_map)), + 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. +/// +/// A ConfigMap that was never applied carries no resource version, in which case the hash covers +/// nothing. +fn discovery_hash(discovery_config_map: &ConfigMap) -> 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(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() +} + +#[cfg(test)] +mod tests { + use stackable_operator::kube::api::ObjectMeta; + + use super::{ConfigMap, discovery_hash}; + + fn discovery_config_map(resource_version: Option<&str>) -> ConfigMap { + ConfigMap { + metadata: ObjectMeta { + name: Some("simple-zookeeper".to_owned()), + resource_version: resource_version.map(ToOwned::to_owned), + ..ObjectMeta::default() + }, + ..ConfigMap::default() + } + } + + /// The hash exists so that clients can tell when the published connection details changed, so + /// it must follow the discovery ConfigMap's resource version. + #[test] + fn discovery_hash_tracks_the_resource_version() { + let hash = discovery_hash(&discovery_config_map(Some("42"))); + + assert_ne!(hash, discovery_hash(&discovery_config_map(Some("43")))); + assert_eq!(hash, discovery_hash(&discovery_config_map(Some("42")))); + + // A ConfigMap that was never applied carries no resource version, in which case the hash + // covers nothing and stays at the hasher's initial state. + assert_ne!(hash, discovery_hash(&discovery_config_map(None))); + } +} diff --git a/rust/operator-binary/src/zk_controller/validate.rs b/rust/operator-binary/src/zk_controller/validate.rs index 93300a46..79e16f01 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,13 @@ 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. + /// + /// Empty until the listener operator has published them, in which case the discovery ConfigMap + /// advertises an empty connection string until the reconciliation that the Listener watch + /// triggers fills it in. + pub discovery_addresses: ListenerAddresses, } // Placeholder product version used for labels on PVC templates, which cannot be modified once @@ -264,6 +275,7 @@ impl ValidatedCluster { >, cluster_operation: ClusterOperation, object_overrides: ObjectOverrides, + discovery_addresses: ListenerAddresses, ) -> Self { Self { metadata: ObjectMeta { @@ -282,6 +294,7 @@ impl ValidatedCluster { role_group_configs, cluster_operation, object_overrides, + discovery_addresses, } } @@ -508,6 +521,19 @@ 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. Both cases resolve to no addresses + // rather than an error, so that the discovery ConfigMap is still built (see + // [`build_discovery_configmap`](crate::discovery::build_discovery_configmap)). + let discovery_addresses = dereferenced_objects + .maybe_role_listener + .as_ref() + .map(|listener| listener_addresses(listener, ZOOKEEPER_SERVER_PORT_NAME)) + .transpose() + .context(ReadRoleListenerAddressesSnafu)? + .flatten() + .unwrap_or_default(); + Ok(ValidatedCluster::new( name, namespace, @@ -522,6 +548,7 @@ pub fn validate( role_group_configs, zk.spec.cluster_operation.clone(), zk.spec.object_overrides.clone(), + discovery_addresses, )) } @@ -577,10 +604,28 @@ mod tests { use stackable_operator::k8s_openapi::apimachinery::pkg::api::resource::Quantity; use super::*; - use crate::zk_controller::test_support::{ - app_version_label, minimal_zk, try_validate, validated_cluster, + use crate::{ + listener_addresses::test_support::{ingress_address, role_listener}, + zk_controller::test_support::{ + app_version_label, minimal_zk, try_validate, try_validate_with_role_listener, + validated_cluster, + }, }; + const MINIMAL_ZK_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + "#; + /// Locks every value the validate step itself derives from the minimal fixture — so a /// validation regression fails here, with a validate-shaped message, instead of surfacing as /// a confusing build-test failure downstream. @@ -798,4 +843,55 @@ mod tests { let _: RoleName = role.into(); } } + + /// The `zk` port is a constant, so a role Listener that publishes addresses without it is a + /// fault rather than a transient state, and must fail validation. + #[test] + fn role_listener_without_the_expected_port_fails_validation() { + let zk = minimal_zk(MINIMAL_ZK_YAML); + let listener = role_listener(Some(vec![ingress_address( + "node-0", + "not-the-zk-port", + 2181, + )])); + + assert!(matches!( + try_validate_with_role_listener(&zk, Some(listener)), + Err(Error::ReadRoleListenerAddresses { .. }) + )); + } + + /// The published addresses reach the discovery ConfigMap through `discovery_addresses`. + #[test] + fn role_listener_addresses_are_carried_into_the_validated_cluster() { + let zk = minimal_zk(MINIMAL_ZK_YAML); + let listener = role_listener(Some(vec![ + ingress_address("node-1", ZOOKEEPER_SERVER_PORT_NAME, 2181), + ingress_address("node-0", ZOOKEEPER_SERVER_PORT_NAME, 2181), + ])); + + let validated = try_validate_with_role_listener(&zk, Some(listener)) + .expect("validate should succeed for the test fixture"); + + assert_eq!( + validated.discovery_addresses.to_connection_string(), + "node-0:2181,node-1:2181" + ); + } + + /// Unlike the znode controller, missing addresses are not an error here: the reconciliation + /// that creates the role Listener in the first place has to get as far as the apply step. + #[test] + fn missing_role_listener_addresses_validate_to_no_addresses() { + let zk = minimal_zk(MINIMAL_ZK_YAML); + + // The Listener does not exist yet. + let validated = validated_cluster(&zk); + assert_eq!(validated.discovery_addresses, ListenerAddresses::default()); + + // The Listener exists, but the listener operator has not published its status yet. + let validated = try_validate_with_role_listener(&zk, Some(role_listener(None))) + .expect("validate should succeed for the test fixture"); + assert_eq!(validated.discovery_addresses, ListenerAddresses::default()); + } } diff --git a/rust/operator-binary/src/znode_controller.rs b/rust/operator-binary/src/znode_controller.rs index c0905988..f9014701 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)?; - 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), - })?; + // build (no client required) + let resources = build::build(validated_znode, znode_path).context(BuildResourcesSnafu)?; - 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()) } @@ -380,6 +330,94 @@ pub fn error_policy( controller::Action::requeue(*Duration::from_secs(5)) } +/// Shared helpers for building validated test znodes from minimal YAML fixtures. +#[cfg(test)] +pub(crate) mod test_support { + use stackable_operator::crd::listener; + + use crate::{ + crd::{authentication::DereferencedAuthenticationClasses, v1alpha1}, + zk_controller::test_support::{minimal_zk, operator_environment}, + znode_controller::{ + dereference::DereferencedObjects, + validate::{ValidatedZnode, validate}, + }, + }; + + /// Parses a minimal `ZookeeperZnode` test fixture, defaulting `namespace`/`uid` so the validate + /// step can build a [`ValidatedZnode`]. + pub fn minimal_znode(yaml: &str) -> v1alpha1::ZookeeperZnode { + let mut znode: v1alpha1::ZookeeperZnode = + serde_yaml::from_str(yaml).expect("invalid test ZookeeperZnode YAML"); + znode + .metadata + .namespace + .get_or_insert_with(|| "default".to_owned()); + znode + .metadata + .uid + .get_or_insert_with(|| "e5dbf9c2-d8b0-4c1e-9f4a-1d2e3f4a5b6c".to_owned()); + znode + } + + /// The `ZookeeperCluster` that the znode fixtures reference. The znode validate step reads the + /// image, security settings and cluster operation from it. + pub fn referenced_zk() -> v1alpha1::ZookeeperCluster { + minimal_zk( + r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + "#, + ) + } + + /// Runs the real validate step against a minimal (auth-free) fixture and the referenced + /// cluster's role Listener, returning the result so tests can assert on validation errors. + pub fn try_validate( + znode: &v1alpha1::ZookeeperZnode, + maybe_role_listener: Option, + ) -> Result { + validate( + znode, + &DereferencedObjects { + zk: referenced_zk(), + authentication_classes: DereferencedAuthenticationClasses::new_for_tests(), + maybe_role_listener, + }, + &operator_environment(), + ) + } + + /// Runs the real validate step against a minimal (auth-free) fixture whose role Listener + /// publishes `node-0:2282`, the secure client port that the referenced cluster serves because + /// its fixture keeps TLS enabled. + pub fn validated_znode(znode: &v1alpha1::ZookeeperZnode) -> ValidatedZnode { + use crate::{ + crd::ZOOKEEPER_SERVER_PORT_NAME, + listener_addresses::test_support::{ingress_address, role_listener}, + }; + + try_validate( + znode, + Some(role_listener(Some(vec![ingress_address( + "node-0", + ZOOKEEPER_SERVER_PORT_NAME, + 2282, + )]))), + ) + .expect("validate should succeed for the test fixture") + } +} + mod znode_mgmt { use std::{collections::VecDeque, net::SocketAddr}; 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..86c9d69d --- /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`] 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..862f3c90 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,86 @@ pub fn validate( zookeeper_security, cluster_operation: dereferenced_objects.zk.spec.cluster_operation.clone(), object_overrides: znode.spec.object_overrides.clone(), + discovery_addresses, }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + listener_addresses::test_support::{ingress_address, role_listener}, + zk_controller::test_support::app_version_label, + znode_controller::test_support::{minimal_znode, try_validate, validated_znode}, + }; + + const ZNODE_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperZnode + metadata: + name: simple-znode + spec: + clusterRef: + name: simple-zookeeper + "#; + + /// Locks the values the validate step derives from the znode itself and from the referenced + /// cluster, including the addresses that the znode's discovery ConfigMap advertises. + #[test] + fn validate_ok_derives_expected_values() { + let validated = validated_znode(&minimal_znode(ZNODE_YAML)); + + assert_eq!(validated.name, "simple-znode"); + assert_eq!(validated.namespace.to_string(), "default"); + assert_eq!( + validated.uid.to_string(), + "e5dbf9c2-d8b0-4c1e-9f4a-1d2e3f4a5b6c" + ); + // The product version comes from the referenced cluster, not from the znode. + assert_eq!( + validated.product_version.to_string(), + app_version_label("3.9.5") + ); + assert!(validated.zookeeper_security.tls_enabled()); + assert_eq!( + validated.discovery_addresses.to_connection_string(), + "node-0:2282" + ); + } + + /// The znode's discovery ConfigMap is the only resource this controller produces, so a role + /// Listener that publishes addresses the znode cannot use must fail validation rather than + /// advertise nothing. + #[test] + fn role_listener_without_the_expected_port_fails_validation() { + let listener = role_listener(Some(vec![ingress_address( + "node-0", + "not-the-zk-port", + 2181, + )])); + + assert!(matches!( + try_validate(&minimal_znode(ZNODE_YAML), Some(listener)), + Err(Error::ReadRoleListenerAddresses { .. }) + )); + } + + /// The znode controller runs on its own schedule, so it can observe the referenced cluster + /// before the cluster controller has created the role Listener at all. + #[test] + fn missing_role_listener_fails_validation() { + assert!(matches!( + try_validate(&minimal_znode(ZNODE_YAML), None), + Err(Error::NoRoleListenerAddresses) + )); + } + + /// The Listener exists, but the listener operator has not published its addresses yet. + #[test] + fn role_listener_without_addresses_fails_validation() { + assert!(matches!( + try_validate(&minimal_znode(ZNODE_YAML), Some(role_listener(None))), + Err(Error::NoRoleListenerAddresses) + )); + } +}