Skip to content

Commit 02b9fd8

Browse files
committed
refactor: Introduce build aggregator
1 parent 5739680 commit 02b9fd8

2 files changed

Lines changed: 242 additions & 119 deletions

File tree

rust/operator-binary/src/controller.rs

Lines changed: 80 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,11 @@ use stackable_operator::{
2121
},
2222
crd::{listener::v1alpha1::Listener, s3},
2323
database_connections::drivers::jdbc::JdbcDatabaseConnectionDetails,
24+
k8s_openapi::api::{
25+
apps::v1::StatefulSet,
26+
core::v1::{ConfigMap, Service},
27+
policy::v1::PodDisruptionBudget,
28+
},
2429
kube::{
2530
Resource, ResourceExt,
2631
api::ObjectMeta,
@@ -49,12 +54,7 @@ use strum::EnumDiscriminants;
4954

5055
use crate::{
5156
OPERATOR_NAME,
52-
controller::build::resource::{
53-
discovery,
54-
listener::build_role_listener,
55-
pdb::build_pdb,
56-
service::{build_rolegroup_headless_service, build_rolegroup_metrics_service},
57-
},
57+
controller::build::resource::discovery,
5858
crd::{APP_NAME, HdfsConnection, HiveClusterStatus, HiveRole, MetaStoreConfig, v1alpha1},
5959
};
6060

@@ -70,28 +70,12 @@ pub struct Ctx {
7070
#[strum_discriminants(derive(strum::IntoStaticStr))]
7171
#[allow(clippy::enum_variant_names)]
7272
pub enum Error {
73-
#[snafu(display("failed to apply Service for role group {role_group}"))]
74-
ApplyRoleGroupService {
75-
source: stackable_operator::cluster_resources::Error,
76-
role_group: RoleGroupName,
77-
},
78-
79-
#[snafu(display("failed to build ConfigMap for role group {role_group}"))]
80-
BuildRoleGroupConfigMap {
81-
source: build::resource::config_map::Error,
82-
role_group: RoleGroupName,
83-
},
73+
#[snafu(display("failed to build the Kubernetes resources"))]
74+
BuildResources { source: build::Error },
8475

85-
#[snafu(display("failed to apply ConfigMap for role group {role_group}"))]
86-
ApplyRoleGroupConfig {
76+
#[snafu(display("failed to apply Kubernetes resource"))]
77+
ApplyResource {
8778
source: stackable_operator::cluster_resources::Error,
88-
role_group: RoleGroupName,
89-
},
90-
91-
#[snafu(display("failed to apply StatefulSet for role group {role_group}"))]
92-
ApplyRoleGroupStatefulSet {
93-
source: stackable_operator::cluster_resources::Error,
94-
role_group: RoleGroupName,
9579
},
9680

9781
#[snafu(display("failed to build discovery ConfigMap"))]
@@ -127,11 +111,6 @@ pub enum Error {
127111
source: stackable_operator::commons::rbac::Error,
128112
},
129113

130-
#[snafu(display("failed to apply PodDisruptionBudget"))]
131-
ApplyPdb {
132-
source: stackable_operator::cluster_resources::Error,
133-
},
134-
135114
#[snafu(display("failed to get required Labels"))]
136115
GetRequiredLabels {
137116
source:
@@ -143,24 +122,13 @@ pub enum Error {
143122
source: error_boundary::InvalidObject,
144123
},
145124

146-
#[snafu(display("failed to apply group listener for {role}"))]
147-
ApplyGroupListener {
148-
source: stackable_operator::cluster_resources::Error,
149-
role: String,
150-
},
151125
#[snafu(display("failed to dereference cluster resources"))]
152126
Dereference {
153127
source: crate::controller::dereference::Error,
154128
},
155129

156130
#[snafu(display("failed to validate cluster configuration"))]
157131
Validate { source: validate::Error },
158-
159-
#[snafu(display("failed to build StatefulSet for role group {role_group}"))]
160-
BuildRoleGroupStatefulSet {
161-
source: build::resource::statefulset::Error,
162-
role_group: RoleGroupName,
163-
},
164132
}
165133
type Result<T, E = Error> = std::result::Result<T, E>;
166134

@@ -462,6 +430,19 @@ pub struct ValidatedRoleConfig {
462430
pub listener_class: ListenerClassName,
463431
}
464432

433+
/// Every Kubernetes resource produced by the client-free [`build`](build::build) step.
434+
///
435+
/// The role-level discovery `ConfigMap` is deliberately absent: it is built from the *applied*
436+
/// role [`Listener`]'s ingress addresses, so it is assembled in the reconcile step after the
437+
/// Listener has been applied, not here.
438+
pub struct KubernetesResources {
439+
pub stateful_sets: Vec<StatefulSet>,
440+
pub services: Vec<Service>,
441+
pub listeners: Vec<Listener>,
442+
pub config_maps: Vec<ConfigMap>,
443+
pub pod_disruption_budgets: Vec<PodDisruptionBudget>,
444+
}
445+
465446
pub async fn reconcile_hive(
466447
hive: Arc<DeserializeGuard<v1alpha1::HiveCluster>>,
467448
ctx: Arc<Ctx>,
@@ -515,98 +496,79 @@ pub async fn reconcile_hive(
515496
.await
516497
.context(ApplyRoleBindingSnafu)?;
517498

518-
let mut ss_cond_builder = StatefulSetConditionBuilder::default();
499+
// The ServiceAccount name is deterministic on the built object, so the client-free build step
500+
// does not depend on the applied ServiceAccount.
501+
let service_account_name = rbac_sa.name_any();
519502

520-
for (hive_role, role_group_configs) in &validated_cluster.role_group_configs {
521-
for (role_group_name, rg) in role_group_configs {
522-
let rg_metrics_service =
523-
build_rolegroup_metrics_service(&validated_cluster, role_group_name);
503+
let resources = build::build(
504+
&validated_cluster,
505+
&client.kubernetes_cluster_info,
506+
&service_account_name,
507+
)
508+
.context(BuildResourcesSnafu)?;
524509

525-
let rg_headless_service =
526-
build_rolegroup_headless_service(&validated_cluster, role_group_name);
510+
let mut ss_cond_builder = StatefulSetConditionBuilder::default();
527511

528-
let rg_configmap = build::resource::config_map::build_metastore_rolegroup_config_map(
529-
&validated_cluster,
530-
&client.kubernetes_cluster_info,
531-
role_group_name,
532-
rg,
533-
)
534-
.with_context(|_| BuildRoleGroupConfigMapSnafu {
535-
role_group: role_group_name.clone(),
536-
})?;
537-
538-
let rg_statefulset =
539-
build::resource::statefulset::build_metastore_rolegroup_statefulset(
540-
hive_role,
541-
&validated_cluster,
542-
role_group_name,
543-
rg,
544-
&rbac_sa.name_any(),
545-
)
546-
.with_context(|_| BuildRoleGroupStatefulSetSnafu {
547-
role_group: role_group_name.clone(),
548-
})?;
512+
// Apply order: everything before StatefulSets, StatefulSets last. A StatefulSet must only be
513+
// applied after all ConfigMaps and Secrets it mounts, to prevent unnecessary Pod restarts.
514+
// See https://github.com/stackabletech/commons-operator/issues/111 for details.
515+
for service in resources.services {
516+
cluster_resources
517+
.add(client, service)
518+
.await
519+
.context(ApplyResourceSnafu)?;
520+
}
549521

522+
// The role Listener is applied before the discovery ConfigMap, which is built below from the
523+
// applied Listener's ingress addresses. Hive has a single role Listener, so at most one is
524+
// captured here.
525+
let mut applied_role_listener: Option<Listener> = None;
526+
for listener in resources.listeners {
527+
applied_role_listener = Some(
550528
cluster_resources
551-
.add(client, rg_metrics_service)
529+
.add(client, listener)
552530
.await
553-
.context(ApplyRoleGroupServiceSnafu {
554-
role_group: role_group_name.clone(),
555-
})?;
531+
.context(ApplyResourceSnafu)?,
532+
);
533+
}
534+
535+
for config_map in resources.config_maps {
536+
cluster_resources
537+
.add(client, config_map)
538+
.await
539+
.context(ApplyResourceSnafu)?;
540+
}
556541

542+
for pdb in resources.pod_disruption_budgets {
543+
cluster_resources
544+
.add(client, pdb)
545+
.await
546+
.context(ApplyResourceSnafu)?;
547+
}
548+
549+
for statefulset in resources.stateful_sets {
550+
ss_cond_builder.add(
557551
cluster_resources
558-
.add(client, rg_headless_service)
552+
.add(client, statefulset)
559553
.await
560-
.context(ApplyRoleGroupServiceSnafu {
561-
role_group: role_group_name.clone(),
562-
})?;
563-
564-
cluster_resources.add(client, rg_configmap).await.context(
565-
ApplyRoleGroupConfigSnafu {
566-
role_group: role_group_name.clone(),
567-
},
568-
)?;
569-
570-
// Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it
571-
// mounts to prevent unnecessary Pod restarts.
572-
// See https://github.com/stackabletech/commons-operator/issues/111 for details.
573-
ss_cond_builder.add(
574-
cluster_resources
575-
.add(client, rg_statefulset)
576-
.await
577-
.context(ApplyRoleGroupStatefulSetSnafu {
578-
role_group: role_group_name.clone(),
579-
})?,
580-
);
581-
}
554+
.context(ApplyResourceSnafu)?,
555+
);
582556
}
583557

558+
// The discovery ConfigMap is built from the *applied* role Listener's ingress addresses, so it
559+
// is assembled here rather than in the client-free build step. Its applied resource version
560+
// feeds the status discovery hash.
584561
// std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases.
585562
// We don't /need/ stability, but it's still nice to avoid spurious changes where possible.
586563
let mut discovery_hash = FnvHasher::with_key(0);
587564

588-
if let Some(role_config) = &validated_cluster.role_config {
589-
if let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, &HiveRole::MetaStore) {
590-
cluster_resources
591-
.add(client, pdb)
592-
.await
593-
.context(ApplyPdbSnafu)?;
594-
}
595-
596-
let role_listener: Listener = build_role_listener(
565+
if let Some(role_listener) = applied_role_listener {
566+
let discovery_cm = discovery::build_discovery_configmap(
597567
&validated_cluster,
598-
&HiveRole::MetaStore,
599-
&role_config.listener_class,
600-
);
601-
let listener = cluster_resources.add(client, role_listener).await.context(
602-
ApplyGroupListenerSnafu {
603-
role: HiveRole::MetaStore.to_string(),
604-
},
605-
)?;
606-
607-
let discovery_cm =
608-
discovery::build_discovery_configmap(&validated_cluster, HiveRole::MetaStore, listener)
609-
.context(BuildDiscoveryConfigSnafu)?;
568+
HiveRole::MetaStore,
569+
role_listener,
570+
)
571+
.context(BuildDiscoveryConfigSnafu)?;
610572
let discovery_cm = cluster_resources
611573
.add(client, discovery_cm)
612574
.await

0 commit comments

Comments
 (0)