diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a4dfcb7..78d5596b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,13 @@ All notable changes to this project will be documented in this file. - The RBAC ServiceAccount and RoleBinding are now built with the operator-rs `v2::rbac` functions and carry the full set of recommended labels ([#846]). - Bump stackable-operator to 0.114.0 ([#855]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps ([#856]). [#841]: https://github.com/stackabletech/druid-operator/pull/841 [#846]: https://github.com/stackabletech/druid-operator/pull/846 [#855]: https://github.com/stackabletech/druid-operator/pull/855 +[#856]: https://github.com/stackabletech/druid-operator/pull/856 ## [26.7.0] - 2026-07-21 diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 642a31bb..2c1c74de 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -1,7 +1,7 @@ //! Ensures that `Pod`s are configured and running for each [`DruidCluster`][v1alpha1] //! //! [v1alpha1]: v1alpha1::DruidCluster -use std::{str::FromStr, sync::Arc}; +use std::{marker::PhantomData, str::FromStr, sync::Arc}; use const_format::concatcp; use snafu::{ResultExt, Snafu}; @@ -21,25 +21,23 @@ 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, OperatorName, ProductName}, - }, + v2::types::operator::{ControllerName, OperatorName, ProductName}, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ - controller::build::resource::listener::{build_group_listener, group_listener_name}, - crd::{APP_NAME, DruidClusterStatus, DruidRole, OPERATOR_NAME, v1alpha1}, - internal_secret::create_shared_internal_secret, + controller::{ + apply::{Applier, ensure_internal_secret}, + build::resource::listener::group_listener_name, + update_status::update_status, + }, + crd::{APP_NAME, DruidRole, OPERATOR_NAME, v1alpha1}, }; +mod apply; mod build; mod dereference; +mod update_status; pub(crate) mod validate; use build::resource::discovery::{self, build_discovery_configmaps}; @@ -70,12 +68,22 @@ pub struct Ctx { pub operator_environment: OperatorEnvironmentOptions, } -/// Every Kubernetes resource produced by the client-free [`build`](build::build) step. +/// 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 build step. +/// +/// The discovery `ConfigMap`s are intentionally *not* part of this bundle: they derive from the +/// *applied* Router listener's ingress address, so they are built after the first apply phase +/// and applied through the same [`apply::Applier`] before its orphan deletion runs. /// -/// The Router group `Listener` and the discovery `ConfigMap`s are intentionally *not* yet part of this -/// bundle: the discovery `ConfigMap` derives from the *applied* Router listener's ingress address, -/// so both are built and applied in the reconcile step instead. -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, @@ -83,19 +91,32 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, +} + +impl KubernetesResources { + /// The applied group [`Listener`] of the given role, if the role exposes one. + pub fn group_listener( + &self, + cluster: &validate::ValidatedCluster, + role: &DruidRole, + ) -> Option<&Listener> { + let listener_name = group_listener_name(cluster, role)?; + self.listeners + .iter() + .find(|listener| listener.metadata.name.as_deref() == Some(listener_name.as_ref())) + } } #[derive(Snafu, Debug, EnumDiscriminants)] #[strum_discriminants(derive(IntoStaticStr))] pub enum Error { + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::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("failed to dereference cluster objects"))] Dereference { source: dereference::Error }, @@ -103,35 +124,22 @@ pub enum Error { BuildDiscoveryConfig { source: discovery::Error }, #[snafu(display("failed to apply discovery ConfigMap"))] - ApplyDiscoveryConfig { - source: stackable_operator::cluster_resources::Error, - }, + ApplyDiscoveryConfig { source: apply::Error }, - #[snafu(display("failed to apply cluster status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphanedResources { - source: stackable_operator::cluster_resources::Error, - }, + DeleteOrphanedResources { source: apply::Error }, #[snafu(display("failed to retrieve secret for internal communications"))] - FailedInternalSecretCreation { - source: crate::internal_secret::Error, - }, + FailedInternalSecretCreation { source: apply::Error }, #[snafu(display("DruidCluster object is invalid"))] InvalidDruidCluster { source: error_boundary::InvalidObject, }, - #[snafu(display("failed to apply group listener"))] - ApplyGroupListener { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to validate cluster"))] ValidateCluster { source: validate::Error }, } @@ -165,124 +173,46 @@ pub async fn reconcile_druid( validate::validate(druid, &dereferenced_objects, &ctx.operator_environment) .context(ValidateClusterSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&druid.spec.cluster_operation), - &druid.spec.object_overrides, - ); + let resources = build::build(&validated_cluster).context(BuildResourcesSnafu)?; - // The internal secret is shared across all roles and role groups, so it only needs to be - // created once per reconcile rather than inside the role loop below. - create_shared_internal_secret(&validated_cluster, client, DRUID_CONTROLLER_NAME) + ensure_internal_secret(client, &validated_cluster) .await .context(FailedInternalSecretCreationSnafu)?; - let resources = build::build(&validated_cluster).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)?; - } - - // Apply order: everything a Pod mounts (ConfigMaps) must exist before the StatefulSets, so the - // StatefulSets are applied last to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - for service in resources.services { - cluster_resources - .add(client, service) - .await - .context(ApplyResourceSnafu)?; - } - for listener in resources.listeners { - cluster_resources - .add(client, listener) - .await - .context(ApplyResourceSnafu)?; - } - for config_map in resources.config_maps { - cluster_resources - .add(client, config_map) - .await - .context(ApplyResourceSnafu)?; - } - for pdb in resources.pod_disruption_budgets { - cluster_resources - .add(client, pdb) - .await - .context(ApplyResourceSnafu)?; - } - for stateful_set in resources.stateful_sets { - ss_cond_builder.add( - cluster_resources - .add(client, stateful_set) - .await - .context(ApplyResourceSnafu)?, - ); - } - - // The Router group Listener and its discovery ConfigMaps are applied here rather than in the - // build step: the discovery ConfigMap derives from the *applied* Router listener's ingress - // address, which is only known after the Listener has been applied. - if let Some(listener_class) = &validated_cluster - .role_config(&DruidRole::Router) - .listener_class - && let Some(listener_group_name) = - group_listener_name(&validated_cluster, &DruidRole::Router) - { - let router_listener = build_group_listener( - &validated_cluster, - listener_class, - listener_group_name, - &DruidRole::Router, - ); - - let listener = cluster_resources - .add(client, router_listener) - .await - .context(ApplyGroupListenerSnafu)?; + let mut applier = Applier::new( + client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&druid.spec.cluster_operation), + &druid.spec.object_overrides, + ); - for discovery_cm in build_discovery_configmaps(&validated_cluster, listener) + let applied = applier + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; + + // Second apply phase: the discovery ConfigMaps derive from the *applied* Router listener's + // ingress address, which is only known after the Listener has been applied. The Router + // listener itself is built in `build()` and applied with all the other listeners in the + // first apply phase above; here it is only read back. The discovery ConfigMaps must go + // through the same Applier, so that the orphan deletion in `finish` sees them. + if let Some(router_listener) = applied.group_listener(&validated_cluster, &DruidRole::Router) { + let discovery_config_maps = build_discovery_configmaps(&validated_cluster, router_listener) + .context(BuildDiscoveryConfigSnafu)?; + applier + .apply_config_maps(discovery_config_maps) .await - .context(BuildDiscoveryConfigSnafu)? - { - cluster_resources - .add(client, discovery_cm) - .await - .context(ApplyDiscoveryConfigSnafu)?; - } + .context(ApplyDiscoveryConfigSnafu)?; } - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&druid.spec.cluster_operation); - - let status = DruidClusterStatus { - conditions: compute_conditions(druid, &[&ss_cond_builder, &cluster_operation_cond_builder]), - }; - - cluster_resources - .delete_orphaned_resources(client) + applier + .finish() .await .context(DeleteOrphanedResourcesSnafu)?; - client - .apply_patch_status(OPERATOR_NAME, druid, &status) + + update_status(client, druid, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } @@ -313,6 +243,42 @@ mod test { crd::PROP_SEGMENT_CACHE_LOCATIONS, }; + /// `group_listener` finds the applied group Listener of a role by its name, and yields + /// `None` for roles that expose no Listener (here: Historical). + #[test] + fn group_listener_finds_the_role_listener_by_name() { + let druid = crate::controller::validate::test_support::druid_from_yaml( + crate::controller::validate::test_support::MINIMAL_DRUID_YAML, + ); + let cluster = crate::controller::validate::test_support::validated_cluster(&druid); + let built = build::build(&cluster).expect("the test cluster builds"); + // `KubernetesResources` normally only exists after an apply; constructing it + // from the built resources is fine here because the lookup only reads names. + let applied = KubernetesResources:: { + stateful_sets: built.stateful_sets, + services: built.services, + listeners: built.listeners, + config_maps: built.config_maps, + pod_disruption_budgets: built.pod_disruption_budgets, + service_accounts: built.service_accounts, + role_bindings: built.role_bindings, + status: std::marker::PhantomData, + }; + + let router_listener = applied + .group_listener(&cluster, &DruidRole::Router) + .expect("the Router exposes a group Listener"); + assert_eq!( + router_listener.metadata.name.as_deref(), + Some("simple-druid-router") + ); + assert!( + applied + .group_listener(&cluster, &DruidRole::Historical) + .is_none() + ); + } + #[rstest] #[case( "segment_cache.yaml", diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..af8407c8 --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,305 @@ +//! The apply step in the DruidCluster controller. + +use std::{ + collections::{BTreeMap, HashSet}, + marker::PhantomData, +}; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + builder::meta::ObjectMetaBuilder, + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + k8s_openapi::api::core::v1::{ConfigMap, Secret}, + kube::ResourceExt, + v2::{builder::meta::ownerreference_from_resource, cluster_resources::cluster_resources_new}, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{ + Applied, KubernetesResources, Prepared, controller_name, operator_name, product_name, + validate::ValidatedCluster, + }, + crd::{COOKIE_PASSPHRASE_ENV, security::INTERNAL_INITIAL_CLIENT_PASSWORD_ENV}, + internal_secret::build_shared_internal_secret_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, + }, + + #[snafu(display("failed to apply internal secret"))] + ApplyInternalSecret { + source: stackable_operator::client::Error, + }, + + #[snafu(display("failed to delete the immutable internal secret"))] + DeleteImmutableInternalSecret { + source: stackable_operator::client::Error, + }, + + #[snafu(display("failed to retrieve secret for internal communications"))] + FailedToRetrieveInternalSecret { + source: stackable_operator::client::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> { + let cluster_resources = cluster_resources_new( + &product_name(), + &operator_name(), + &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. + /// + /// Resources derived from the applied state (the discovery `ConfigMap`s) can be applied + /// afterwards via [`Self::apply_config_maps`]; [`Self::finish`] must be called once all + /// resources are applied, so that orphaned resources are deleted exactly once at the end. + 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, + 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 -- 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 pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + let stateful_sets = self.add_resources(stateful_sets).await?; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }) + } + + /// Applies `ConfigMap`s that are derived from already-applied resources (the discovery + /// `ConfigMap`s, which need the applied Router listener's address). + pub async fn apply_config_maps( + &mut self, + config_maps: Vec, + ) -> Result> { + self.add_resources(config_maps).await + } + + /// Deletes resources from earlier reconcile runs that were not applied in this one. + /// + /// Must be called exactly once, after every apply phase: a resource applied after this call + /// would be treated as an orphan and deleted by the next reconcile run. + pub async fn finish(self) -> Result<()> { + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + let applied_resource = self + .cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu)?; + applied_resources.push(applied_resource); + } + + Ok(applied_resources) + } +} + +/// Ensures the shared internal Secret (cookie passphrase + initial client password) exists, +/// creating it when missing, migrating away from the pre-2024-06 immutable Secret, and patching +/// in newly required keys. These are read-then-write client operations, so they cannot be part +/// of the client-free `build()` step; the Secret is also deliberately not tracked in +/// [`ClusterResources`], so it survives orphan deletion. Unlike a create-if-absent Secret, the +/// migration and repair paths may rewrite the contents (documented downtime warnings below). +pub async fn ensure_internal_secret(client: &Client, cluster: &ValidatedCluster) -> Result<()> { + let controller_name = controller_name(); + let controller_name = controller_name.as_ref(); + let secret = build_shared_internal_secret(cluster); + let existing_secret = client + .get_opt::(&secret.name_any(), cluster.namespace.as_ref()) + .await + .context(FailedToRetrieveInternalSecretSnafu)?; + let existing_immutable_secret = client + .get_opt::( + &build_immutable_shared_internal_secret_name(cluster), + cluster.namespace.as_ref(), + ) + .await + .context(FailedToRetrieveInternalSecretSnafu)?; + + match existing_secret { + None => { + match existing_immutable_secret { + None => { + tracing::info!( + secret_name = secret.name_any(), + "Did not found a shared internal secret with the necessary data, creating one" + ); + client + .apply_patch(controller_name, &secret, &secret) + .await + .context(ApplyInternalSecretSnafu)?; + } + Some(existing_immutable_secret) => { + // Before 2024-06-25 we did set `spec.immutable` to avoid accidentally changing the contents. Which was + // great back than, *but* we now need something more flexible. We can not make the Secret mutable, + // and re-creation with the same name is very error-prone so we create a mutable secret with a new name + // (see ). + // We *could* read in the contents and use them during the re-creation (so we don't change the contents to avoid downtime), + // but we strive that our operators don't handle Secret contents and it's a one time migration. + + tracing::warn!( + secret_name = secret.name_any(), + "Shared internal secret found, which is immutable. Re-creating it with a new name, as we can not modify it or re-create it \ + with the same name. This should only happen once and will change the contents of the Secret. This might cause a short \ + downtime of Druid, as the changed internal secrets need to propagate through all Druid nodes" + ); + + client + .delete(&existing_immutable_secret) + .await + .context(DeleteImmutableInternalSecretSnafu)?; + + client + .apply_patch(controller_name, &secret, &secret) + .await + .context(ApplyInternalSecretSnafu)?; + return Ok(()); + } + } + } + + Some(existing_secret) => { + let current_secret_keys = existing_secret + .data + .unwrap_or_default() + .into_keys() + .collect::>(); + for required in INTERNAL_SECRET_KEYS { + if !current_secret_keys.contains(required) { + tracing::info!( + secret_name = secret.name_any(), + "Found shared internal secret, which is missing the key {required}, patching it" + ); + tracing::warn!( + secret_name = secret.name_any(), + "Found shared internal secret, which is missing the key {required}, patching it. This \ + should only happen once and will change the contents of the Secret. This might cause a short \ + downtime of Druid, as the changed internal Secrets need to propagate through all Druid nodes" + ); + client + .apply_patch(controller_name, &secret, &secret) + .await + .context(ApplyInternalSecretSnafu)?; + return Ok(()); + } + } + } + } + + Ok(()) +} + +/// The keys the shared internal Secret must contain. Single source for both +/// [`build_shared_internal_secret`] and the missing-key check in [`ensure_internal_secret`]. +const INTERNAL_SECRET_KEYS: [&str; 2] = + [INTERNAL_INITIAL_CLIENT_PASSWORD_ENV, COOKIE_PASSPHRASE_ENV]; + +fn build_shared_internal_secret(cluster: &ValidatedCluster) -> Secret { + let internal_secret: BTreeMap = INTERNAL_SECRET_KEYS + .iter() + .map(|key| (key.to_string(), get_random_base64())) + .collect(); + + Secret { + metadata: ObjectMetaBuilder::new() + .name(build_shared_internal_secret_name(cluster)) + .namespace_opt(cluster.namespace()) + .ownerreference(ownerreference_from_resource(cluster, None, Some(true))) + .build(), + string_data: Some(internal_secret), + ..Secret::default() + } +} + +fn build_immutable_shared_internal_secret_name(cluster: &ValidatedCluster) -> String { + format!("{}-internal-secret", cluster.name_any()) +} + +fn get_random_base64() -> String { + let mut buf = [0; 512]; + openssl::rand::rand_bytes(&mut buf).expect( + "the OpenSSL CSPRNG could not supply random bytes (e.g. it is not seeded); \ + continuing would turn the zero-initialized buffer into a predictable secret", + ); + openssl::base64::encode_block(&buf) +} diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 10f57a86..9d27be35 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,24 +1,25 @@ //! Build steps that turn a `ValidatedCluster` into Kubernetes resources. -use std::str::FromStr; +use std::{marker::PhantomData, str::FromStr}; use snafu::{ResultExt, Snafu}; -use stackable_operator::v2::types::operator::RoleGroupName; - -use crate::{ - controller::{ - KubernetesResources, - build::resource::{ - config_map::build_rolegroup_config_map, - listener::{build_group_listener, group_listener_name}, - pdb::build_pdb, - rbac::{build_role_binding, build_service_account}, - service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, - statefulset::build_rolegroup_statefulset, - }, - validate::ValidatedCluster, +use stackable_operator::{ + builder::meta::ObjectMetaBuilder, + kvp::Labels, + v2::{builder::meta::ownerreference_from_resource, types::operator::RoleGroupName}, +}; + +use crate::controller::{ + KubernetesResources, Prepared, + build::resource::{ + config_map::build_rolegroup_config_map, + listener::{build_group_listener, group_listener_name}, + pdb::build_pdb, + rbac::{build_role_binding, build_service_account}, + service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, + statefulset::build_rolegroup_statefulset, }, - crd::DruidRole, + validate::ValidatedCluster, }; // Placeholder role-group name used for the recommended labels of the role-level discovery @@ -36,6 +37,25 @@ pub mod properties; pub mod resource; pub mod security; +/// Returns an [`ObjectMetaBuilder`] pre-filled with the cluster's namespace, an owner +/// reference back to the cluster, the resource `name` and the given `recommended_labels`. +/// +/// Consolidates the metadata chain repeated by the child-resource builders. Call sites that +/// need extra labels/annotations chain them onto the returned builder. +pub(crate) fn object_meta( + cluster: &ValidatedCluster, + name: impl Into, + recommended_labels: Labels, +) -> ObjectMetaBuilder { + let mut builder = ObjectMetaBuilder::new(); + builder + .name_and_namespace(cluster) + .name(name) + .ownerreference(ownerreference_from_resource(cluster, None, Some(true))) + .with_labels(recommended_labels); + builder +} + #[derive(Snafu, Debug)] pub enum Error { #[snafu(display("failed to build ConfigMap for role group {role_group}"))] @@ -57,10 +77,10 @@ pub enum Error { /// dereferenced and validated by this point. /// The remaining errors are resource-assembly failures only. /// -/// The Router group `Listener` and the discovery `ConfigMap`s are not built here: the discovery -/// `ConfigMap` derives from the *applied* Router listener's ingress address, so both are built and -/// applied in the reconcile step instead. -pub fn build(cluster: &ValidatedCluster) -> Result { +/// The discovery `ConfigMap`s are not built here: they derive from the *applied* Router +/// listener's ingress address, which is only known after the Listener has been applied. They are +/// built in the reconcile step and applied as a second phase, before orphan deletion. +pub fn build(cluster: &ValidatedCluster) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut listeners = vec![]; @@ -74,10 +94,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result { pod_disruption_budgets.push(pdb); } - // The Router group Listener is built and applied in the reconcile step instead (see the - // module docs and [`KubernetesResources`]), so it is skipped here. - if *druid_role != DruidRole::Router - && let Some(listener_class) = &role_config.listener_class + if let Some(listener_class) = &role_config.listener_class && let Some(listener_group_name) = group_listener_name(cluster, druid_role) { listeners.push(build_group_listener( @@ -124,6 +141,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result { pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } @@ -138,6 +156,18 @@ mod tests { MINIMAL_DRUID_YAML, druid_from_yaml, validated_cluster, }; + /// The expected `app.kubernetes.io/version` label value for the given product version. + /// + /// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main + /// but rewritten by the release process — so tests must derive it rather than hardcode it, + /// or they fail on release branches. + fn app_version_label(product_version: &str) -> String { + format!( + "{product_version}-stackable{}", + crate::built_info::PKG_VERSION + ) + } + fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { let mut names: Vec<&str> = resources .iter() @@ -170,12 +200,15 @@ mod tests { expected_role_group_names ); - // Group Listeners are built for the externally reachable roles except the Router (whose - // Listener is applied in the reconcile step): Broker and Coordinator. - // TODO: add router listener here once properly build in the built step. + // Group Listeners are built for the externally reachable roles: Broker, Coordinator and + // Router. assert_eq!( sorted_names(&resources.listeners), - ["simple-druid-broker", "simple-druid-coordinator"] + [ + "simple-druid-broker", + "simple-druid-coordinator", + "simple-druid-router" + ] ); } @@ -199,18 +232,18 @@ mod tests { let expected_labels = BTreeMap::from( [ - ("app.kubernetes.io/component", "none"), - ("app.kubernetes.io/instance", "simple-druid"), + ("app.kubernetes.io/component", "none".to_string()), + ("app.kubernetes.io/instance", "simple-druid".to_string()), ( "app.kubernetes.io/managed-by", - "druid.stackable.tech_druidcluster", + "druid.stackable.tech_druidcluster".to_string(), ), - ("app.kubernetes.io/name", "druid"), - ("app.kubernetes.io/role-group", "none"), - ("app.kubernetes.io/version", "30.0.0-stackable0.0.0-dev"), - ("stackable.tech/vendor", "Stackable"), + ("app.kubernetes.io/name", "druid".to_string()), + ("app.kubernetes.io/role-group", "none".to_string()), + ("app.kubernetes.io/version", app_version_label("30.0.0")), + ("stackable.tech/vendor", "Stackable".to_string()), ] - .map(|(key, value)| (key.to_string(), value.to_string())), + .map(|(key, value)| (key.to_string(), value)), ); let service_account = resources .service_accounts diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index 6e3c0357..e4eee63e 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -28,6 +28,7 @@ use crate::{ build::{ authentication::generate_runtime_properties_config as generate_auth_runtime_properties, jvm::{AWS_REGION, construct_jvm_args}, + object_meta, properties::{ ConfigFileName, extensions::get_extension_list, @@ -375,13 +376,12 @@ pub fn build_rolegroup_config_map( let mut config_map_builder = ConfigMapBuilder::new(); config_map_builder.metadata( - cluster - .object_meta( - resource_names.role_group_config_map().to_string(), - role, - role_group_name, - ) - .build(), + object_meta( + cluster, + resource_names.role_group_config_map().to_string(), + cluster.recommended_labels(role, role_group_name), + ) + .build(), ); for (filename, file_content) in cm_conf_data.iter() { diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index eca94b62..c65f7702 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -10,7 +10,8 @@ use stackable_operator::{ use crate::{ controller::{ build::{ - PLACEHOLDER_DISCOVERY_ROLE_GROUP, resource::listener::build_listener_connection_string, + PLACEHOLDER_DISCOVERY_ROLE_GROUP, object_meta, + resource::listener::build_listener_connection_string, }, validate::ValidatedCluster, }, @@ -31,9 +32,9 @@ pub enum Error { } /// Builds discovery [`ConfigMap`]s for connecting to a Druid cluster. -pub async fn build_discovery_configmaps( +pub fn build_discovery_configmaps( cluster: &ValidatedCluster, - listener: Listener, + listener: &Listener, ) -> Result, Error> { Ok(vec![build_discovery_configmap(cluster, listener)?]) } @@ -41,7 +42,7 @@ pub async fn build_discovery_configmaps( /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain Druid cluster. fn build_discovery_configmap( cluster: &ValidatedCluster, - listener: Listener, + listener: &Listener, ) -> Result { let router_host = build_listener_connection_string( listener, @@ -57,13 +58,12 @@ fn build_discovery_configmap( ConfigMapBuilder::new() .metadata( - cluster - .object_meta( - cluster.name.to_string(), - &DruidRole::Router, - &PLACEHOLDER_DISCOVERY_ROLE_GROUP, - ) - .build(), + object_meta( + cluster, + cluster.name.to_string(), + cluster.recommended_labels(&DruidRole::Router, &PLACEHOLDER_DISCOVERY_ROLE_GROUP), + ) + .build(), ) .add_data("DRUID_ROUTER", router_host) .add_data("DRUID_SQLALCHEMY", sqlalchemy_conn_str) diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index 78845d01..6716df73 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -15,7 +15,7 @@ use stackable_operator::{ use crate::{ controller::{ - build::{NONE_ROLE_GROUP_NAME, security::listener_ports}, + build::{NONE_ROLE_GROUP_NAME, object_meta, security::listener_ports}, validate::ValidatedCluster, }, crd::{ @@ -48,13 +48,12 @@ pub fn build_group_listener( // The group listener is a role-level (not role-group-level) object, so there is no real // role-group name; the placeholder is used for the recommended labels. Listener { - metadata: cluster - .object_meta( - listener_group_name.to_string(), - druid_role, - &NONE_ROLE_GROUP_NAME, - ) - .build(), + metadata: object_meta( + cluster, + listener_group_name.to_string(), + cluster.recommended_labels(druid_role, &NONE_ROLE_GROUP_NAME), + ) + .build(), spec: listener::v1alpha1::ListenerSpec { class_name: Some(listener_class.to_string()), ports: Some(listener_ports( @@ -96,14 +95,15 @@ pub fn group_listener_name( // Builds the connection string with respect to the listener provided objects pub fn build_listener_connection_string( - listener: Listener, + listener: &Listener, druid_tls_security: &DruidTlsSecurity, role_name: &String, ) -> Result { // We only need the first address corresponding to the role let listener_address = listener .status - .and_then(|s| s.ingress_addresses?.into_iter().next()) + .as_ref() + .and_then(|status| status.ingress_addresses.as_ref()?.first()) .context(RoleListenerHasNoAddressSnafu { role_name })?; let port_name = match druid_tls_security.tls_enabled() { true => TLS_PORT_NAME, @@ -200,7 +200,7 @@ mod tests { fn connection_string_uses_plaintext_port_without_tls() { let tls = DruidTlsSecurity::new(false, None); let conn = build_listener_connection_string( - listener_with_both_ports(), + &listener_with_both_ports(), &tls, &"router".to_string(), ) @@ -212,7 +212,7 @@ mod tests { fn connection_string_uses_tls_port_with_tls() { let tls = DruidTlsSecurity::new(false, Some(SecretClassName::from_str("tls").unwrap())); let conn = build_listener_connection_string( - listener_with_both_ports(), + &listener_with_both_ports(), &tls, &"router".to_string(), ) diff --git a/rust/operator-binary/src/controller/build/resource/service.rs b/rust/operator-binary/src/controller/build/resource/service.rs index f8bfb5c9..cd2bfca4 100644 --- a/rust/operator-binary/src/controller/build/resource/service.rs +++ b/rust/operator-binary/src/controller/build/resource/service.rs @@ -7,7 +7,10 @@ use stackable_operator::{ }; use crate::{ - controller::{build::security::service_ports, validate::ValidatedCluster}, + controller::{ + build::{object_meta, security::service_ports}, + validate::ValidatedCluster, + }, crd::{DruidRole, METRICS_PORT, METRICS_PORT_NAME}, }; @@ -19,16 +22,15 @@ pub fn build_rolegroup_headless_service( role_group_name: &RoleGroupName, ) -> Service { Service { - metadata: cluster - .object_meta( - cluster - .role_group_resource_names(druid_role, role_group_name) - .headless_service_name() - .to_string(), - druid_role, - role_group_name, - ) - .build(), + metadata: object_meta( + cluster, + cluster + .role_group_resource_names(druid_role, role_group_name) + .headless_service_name() + .to_string(), + cluster.recommended_labels(druid_role, role_group_name), + ) + .build(), spec: Some(ServiceSpec { // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), @@ -56,23 +58,22 @@ pub fn build_rolegroup_metrics_service( role_group_name: &RoleGroupName, ) -> Service { Service { - metadata: cluster - .object_meta( - cluster - .role_group_resource_names(druid_role, role_group_name) - .metrics_service_name() - .to_string(), - druid_role, - role_group_name, - ) - .with_labels(prometheus_labels(&Scraping::Enabled)) - .with_annotations(prometheus_annotations( - &Scraping::Enabled, - &Scheme::Http, - "/metrics", - &METRICS_PORT, - )) - .build(), + metadata: object_meta( + cluster, + cluster + .role_group_resource_names(druid_role, role_group_name) + .metrics_service_name() + .to_string(), + cluster.recommended_labels(druid_role, role_group_name), + ) + .with_labels(prometheus_labels(&Scraping::Enabled)) + .with_annotations(prometheus_annotations( + &Scraping::Enabled, + &Scheme::Http, + "/metrics", + &METRICS_PORT, + )) + .build(), spec: Some(ServiceSpec { // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 7c267d63..2f1bc8f4 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -36,6 +36,7 @@ use crate::{ build::{ authentication, graceful_shutdown::add_graceful_shutdown_config, + object_meta, properties::product_logging::MAX_DRUID_LOG_FILES_SIZE, resource::listener::{ LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, build_group_listener_pvc, @@ -344,14 +345,13 @@ pub fn build_rolegroup_statefulset( pod_template.merge_from(rg.pod_overrides.clone()); Ok(StatefulSet { - metadata: cluster - .object_meta( - resource_names.stateful_set_name().to_string(), - role, - role_group_name, - ) - .with_label(RESTART_CONTROLLER_ENABLED_LABEL.to_owned()) - .build(), + metadata: object_meta( + cluster, + resource_names.stateful_set_name().to_string(), + cluster.recommended_labels(role, role_group_name), + ) + .with_label(RESTART_CONTROLLER_ENABLED_LABEL.to_owned()) + .build(), spec: Some(StatefulSetSpec { pod_management_policy: Some("Parallel".to_string()), // Leave `replicas` unset when the role group does not specify a count, so a diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs new file mode 100644 index 00000000..1ee01c06 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,55 @@ +//! The update_status step in the DruidCluster controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{Applied, KubernetesResources}, + crd::{DruidClusterStatus, OPERATOR_NAME, v1alpha1}, +}; + +#[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::DruidCluster`]. Takes [`KubernetesResources`] so the type system +/// proves the status derives from applied resources, not merely built ones. +pub async fn update_status( + client: &Client, + druid: &v1alpha1::DruidCluster, + applied: &KubernetesResources, +) -> Result<()> { + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + ss_cond_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&druid.spec.cluster_operation); + + let status = DruidClusterStatus { + conditions: compute_conditions(druid, &[&ss_cond_builder, &cluster_operation_cond_builder]), + }; + + client + .apply_patch_status(OPERATOR_NAME, druid, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 140a0512..b110f9ed 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -11,7 +11,6 @@ use std::{ use snafu::{ResultExt, Snafu, ensure}; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, cli::OperatorEnvironmentOptions, commons::{ pdb::PdbConfig, @@ -24,7 +23,6 @@ use stackable_operator::{ kvp::Labels, v2::{ HasName, HasUid, NameIsValidLabelValue, - builder::meta::ownerreference_from_resource, controller_utils::{get_cluster_name, get_namespace, get_uid}, kvp::label::{recommended_labels, role_group_selector}, role_group_utils::ResourceNames, @@ -244,26 +242,6 @@ impl ValidatedCluster { role_group_selector(self, &product_name(), &role.into(), role_group_name) } - /// Returns an [`ObjectMetaBuilder`] pre-filled with the namespace, an owner reference back to - /// this cluster, and the recommended labels for a resource named `name` in `role`/`role_group_name`. - /// - /// Consolidates the metadata chain repeated by the child-resource builders. Call sites that need - /// extra labels/annotations chain them onto the returned builder. - pub(crate) fn object_meta( - &self, - name: impl Into, - role: &DruidRole, - role_group_name: &RoleGroupName, - ) -> ObjectMetaBuilder { - let mut builder = ObjectMetaBuilder::new(); - builder - .name_and_namespace(self) - .name(name) - .ownerreference(ownerreference_from_resource(self, None, Some(true))) - .with_labels(self.recommended_labels(role, role_group_name)); - builder - } - /// Type-safe names for the per-cluster RBAC resources: the ServiceAccount shared by all /// Pods, its (namespaced) RoleBinding, and the operator-deployed ClusterRole it binds. pub fn cluster_resource_names(&self) -> role_utils::ResourceNames { diff --git a/rust/operator-binary/src/internal_secret.rs b/rust/operator-binary/src/internal_secret.rs index 3fd8f4ae..1fa1a966 100644 --- a/rust/operator-binary/src/internal_secret.rs +++ b/rust/operator-binary/src/internal_secret.rs @@ -1,179 +1,15 @@ -use std::collections::{BTreeMap, HashSet}; +//! Pure helpers around the shared internal Secret: its name and the env vars that reference +//! it. The Secret itself is ensured by `ensure_internal_secret` in the apply step. -use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, - client::Client, - k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, Secret, SecretKeySelector}, + k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector}, kube::ResourceExt, - v2::builder::meta::ownerreference_from_resource, }; -use crate::{ - controller::validate::ValidatedCluster, - crd::{COOKIE_PASSPHRASE_ENV, security::INTERNAL_INITIAL_CLIENT_PASSWORD_ENV}, -}; - -#[derive(Snafu, Debug)] -#[allow(clippy::enum_variant_names)] -pub enum Error { - #[snafu(display("failed to apply internal secret"))] - ApplyInternalSecret { - source: stackable_operator::client::Error, - }, - - #[snafu(display("failed to delete the immutable internal secret"))] - DeleteImmutableInternalSecret { - source: stackable_operator::client::Error, - }, - - #[snafu(display("failed to retrieve secret for internal communications"))] - FailedToRetrieveInternalSecret { - source: stackable_operator::client::Error, - }, - - #[snafu(display("object defines no namespace"))] - ObjectHasNoNamespace, -} - -pub async fn create_shared_internal_secret( - cluster: &ValidatedCluster, - client: &Client, - controller_name: &str, -) -> Result<(), Error> { - let secret = build_shared_internal_secret(cluster); - let existing_secret = client - .get_opt::( - &secret.name_any(), - secret - .namespace() - .as_deref() - .context(ObjectHasNoNamespaceSnafu)?, - ) - .await - .context(FailedToRetrieveInternalSecretSnafu)?; - let existing_immutable_secret = client - .get_opt::( - &build_immutable_shared_internal_secret_name(cluster), - secret - .namespace() - .as_deref() - .context(ObjectHasNoNamespaceSnafu)?, - ) - .await - .context(FailedToRetrieveInternalSecretSnafu)?; - - match existing_secret { - None => { - match existing_immutable_secret { - None => { - tracing::info!( - secret_name = secret.name_any(), - "Did not found a shared internal secret with the necessary data, creating one" - ); - client - .apply_patch(controller_name, &secret, &secret) - .await - .context(ApplyInternalSecretSnafu)?; - } - Some(existing_immutable_secret) => { - // Before 2024-06-25 we did set `spec.immutable` to avoid accidentally changing the contents. Which was - // great back than, *but* we now need something more flexible. We can not make the Secret mutable, - // and re-creation with the same name is very error-prone so we create a mutable secret with a new name - // (see ). - // We *could* read in the contents and use them during the re-creation (so we don't change the contents to avoid downtime), - // but we strive that our operators don't handle Secret contents and it's a one time migration. - - tracing::warn!( - secret_name = secret.name_any(), - "Shared internal secret found, which is immutable. Re-creating it with a new name, as we can not modify it or re-create it \ - with the same name. This should only happen once and will change the contents of the Secret. This might cause a short \ - downtime of Druid, as the changed internal secrets need to propagate through all Druid nodes" - ); - - client - .delete(&existing_immutable_secret) - .await - .context(DeleteImmutableInternalSecretSnafu)?; - - client - .apply_patch(controller_name, &secret, &secret) - .await - .context(ApplyInternalSecretSnafu)?; - return Ok(()); - } - } - } - - Some(existing_secret) => { - let current_secret_keys = existing_secret - .data - .unwrap_or_default() - .into_keys() - .collect::>(); - for required in secret - .string_data - .as_ref() - .expect("Secret data must be set by the `build_shared_internal_secret` function") - .keys() - { - if !current_secret_keys.contains(required) { - tracing::info!( - secret_name = secret.name_any(), - "Found shared internal secret, which is missing the key {required}, patching it" - ); - tracing::warn!( - secret_name = secret.name_any(), - "Found shared internal secret, which is missing the key {required}, patching it. This \ - should only happen once and will change the contents of the Secret. This might cause a short \ - downtime of Druid, as the changed internal Secrets need to propagate through all Druid nodes" - ); - client - .apply_patch(controller_name, &secret, &secret) - .await - .context(ApplyInternalSecretSnafu)?; - return Ok(()); - } - } - } - } - - Ok(()) -} - -fn build_shared_internal_secret(cluster: &ValidatedCluster) -> Secret { - let mut internal_secret = BTreeMap::new(); - internal_secret.insert( - INTERNAL_INITIAL_CLIENT_PASSWORD_ENV.to_string(), - get_random_base64(), - ); - internal_secret.insert(COOKIE_PASSPHRASE_ENV.to_string(), get_random_base64()); - - Secret { - metadata: ObjectMetaBuilder::new() - .name(build_shared_internal_secret_name(cluster)) - .namespace_opt(cluster.namespace()) - .ownerreference(ownerreference_from_resource(cluster, None, Some(true))) - .build(), - string_data: Some(internal_secret), - ..Secret::default() - } -} - -fn build_immutable_shared_internal_secret_name(cluster: &ValidatedCluster) -> String { - format!("{}-internal-secret", cluster.name_any()) -} - pub fn build_shared_internal_secret_name(owner: &T) -> String { format!("{}-shared-internal-secret", owner.name_any()) } -fn get_random_base64() -> String { - let mut buf = [0; 512]; - openssl::rand::rand_bytes(&mut buf).unwrap(); - openssl::base64::encode_block(&buf) -} - /// Give a secret name and an optional key in the secret to use. /// The value from the key will be set into the given env var name. /// If not secret key is given, the env var name will be used as the secret key.