Skip to content

Commit 3f0502f

Browse files
maltesandermaltesander
committed
refactor: add dereference and validate steps
1 parent 241573c commit 3f0502f

5 files changed

Lines changed: 301 additions & 199 deletions

File tree

‎rust/operator-binary/src/controller.rs‎

Lines changed: 38 additions & 175 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,14 @@
11
//! Ensures that `Pod`s are configured and running for each [`v1alpha1::KafkaCluster`].
22
3-
use std::{collections::HashMap, str::FromStr, sync::Arc};
3+
use std::{str::FromStr, sync::Arc};
44

55
use const_format::concatcp;
6-
use product_config::{ProductConfigManager, types::PropertyNameKind};
6+
use product_config::ProductConfigManager;
77
use snafu::{ResultExt, Snafu};
88
use stackable_operator::{
99
cli::OperatorEnvironmentOptions,
1010
cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
11-
commons::{
12-
product_image_selection::{self},
13-
rbac::build_rbac_resources,
14-
},
11+
commons::rbac::build_rbac_resources,
1512
crd::listener,
1613
kube::{
1714
Resource,
@@ -20,10 +17,6 @@ use stackable_operator::{
2017
runtime::{controller::Action, reflector::ObjectRef},
2118
},
2219
logging::controller::ReconcilerError,
23-
product_config_utils::{
24-
ValidatedRoleConfigByPropertyKind, transform_all_roles_to_config,
25-
validate_all_roles_and_groups_config,
26-
},
2720
role_utils::{GenericRoleConfig, RoleGroupRef},
2821
shared::time::Duration,
2922
status::condition::{
@@ -33,16 +26,14 @@ use stackable_operator::{
3326
};
3427
use strum::{EnumDiscriminants, IntoStaticStr};
3528

29+
mod dereference;
30+
mod validate;
31+
3632
use crate::{
3733
crd::{
38-
self, APP_NAME, CONTAINER_IMAGE_BASE_NAME, JVM_SECURITY_PROPERTIES_FILE,
39-
KafkaClusterStatus, OPERATOR_NAME, authorization,
34+
self, APP_NAME, KafkaClusterStatus, OPERATOR_NAME,
4035
listener::get_kafka_listener_config,
41-
role::{
42-
AnyConfig, KafkaRole, broker::BROKER_PROPERTIES_FILE,
43-
controller::CONTROLLER_PROPERTIES_FILE,
44-
},
45-
security::KafkaTlsSecurity,
36+
role::{AnyConfig, KafkaRole},
4637
v1alpha1,
4738
},
4839
discovery::{self, build_discovery_configmap},
@@ -68,6 +59,12 @@ pub struct Ctx {
6859
#[strum_discriminants(derive(IntoStaticStr))]
6960
#[allow(clippy::enum_variant_names)]
7061
pub enum Error {
62+
#[snafu(display("failed to dereference resources"))]
63+
Dereference { source: dereference::Error },
64+
65+
#[snafu(display("failed to validate cluster"))]
66+
ValidateCluster { source: validate::Error },
67+
7168
#[snafu(display("failed to build pod descriptors"))]
7269
BuildPodDescriptors { source: crate::crd::Error },
7370

@@ -76,12 +73,6 @@ pub enum Error {
7673
source: crate::crd::listener::KafkaListenerError,
7774
},
7875

79-
#[snafu(display("cluster object defines no '{role}' role"))]
80-
MissingKafkaRole {
81-
source: crate::crd::Error,
82-
role: KafkaRole,
83-
},
84-
8576
#[snafu(display("failed to apply role Service"))]
8677
ApplyRoleService {
8778
source: stackable_operator::cluster_resources::Error,
@@ -105,16 +96,6 @@ pub enum Error {
10596
rolegroup: RoleGroupRef<v1alpha1::KafkaCluster>,
10697
},
10798

108-
#[snafu(display("failed to generate product config"))]
109-
GenerateProductConfig {
110-
source: stackable_operator::product_config_utils::Error,
111-
},
112-
113-
#[snafu(display("invalid product config"))]
114-
InvalidProductConfig {
115-
source: stackable_operator::product_config_utils::Error,
116-
},
117-
11899
#[snafu(display("failed to build discovery ConfigMap"))]
119100
BuildDiscoveryConfig { source: discovery::Error },
120101

@@ -128,9 +109,6 @@ pub enum Error {
128109
source: stackable_operator::cluster_resources::Error,
129110
},
130111

131-
#[snafu(display("failed to initialize security context"))]
132-
FailedToInitializeSecurityContext { source: crate::crd::security::Error },
133-
134112
#[snafu(display("failed to create cluster resources"))]
135113
CreateClusterResources {
136114
source: stackable_operator::cluster_resources::Error,
@@ -170,9 +148,6 @@ pub enum Error {
170148
stackable_operator::kvp::KeyValuePairError<stackable_operator::kvp::LabelValueError>,
171149
},
172150

173-
#[snafu(display("failed to validate authentication method"))]
174-
FailedToValidateAuthenticationMethod { source: crate::crd::security::Error },
175-
176151
#[snafu(display("KafkaCluster object is invalid"))]
177152
InvalidKafkaCluster {
178153
source: error_boundary::InvalidObject,
@@ -181,11 +156,6 @@ pub enum Error {
181156
#[snafu(display("KafkaCluster object is misconfigured"))]
182157
MisconfiguredKafkaCluster { source: crd::Error },
183158

184-
#[snafu(display("failed to resolve product image"))]
185-
ResolveProductImage {
186-
source: product_image_selection::Error,
187-
},
188-
189159
#[snafu(display("failed to parse role: {source}"))]
190160
ParseRole { source: strum::ParseError },
191161

@@ -208,9 +178,6 @@ pub enum Error {
208178
BuildListener {
209179
source: crate::resource::listener::Error,
210180
},
211-
212-
#[snafu(display("object defines no namespace"))]
213-
GetOpaConfig { source: authorization::Error },
214181
}
215182
type Result<T, E = Error> = std::result::Result<T, E>;
216183

@@ -221,17 +188,15 @@ impl ReconcilerError for Error {
221188

222189
fn secondary_object(&self) -> Option<ObjectRef<DynamicObject>> {
223190
match self {
224-
Error::MissingKafkaRole { .. } => None,
191+
Error::Dereference { .. } => None,
192+
Error::ValidateCluster { .. } => None,
225193
Error::ApplyRoleService { .. } => None,
226194
Error::ApplyRoleGroupService { .. } => None,
227195
Error::ApplyRoleGroupConfig { .. } => None,
228196
Error::ApplyRoleGroupStatefulSet { .. } => None,
229-
Error::GenerateProductConfig { .. } => None,
230-
Error::InvalidProductConfig { .. } => None,
231197
Error::BuildDiscoveryConfig { .. } => None,
232198
Error::ApplyDiscoveryConfig { .. } => None,
233199
Error::DeleteOrphans { .. } => None,
234-
Error::FailedToInitializeSecurityContext { .. } => None,
235200
Error::CreateClusterResources { .. } => None,
236201
Error::FailedToResolveConfig { .. } => None,
237202
Error::ApplyServiceAccount { .. } => None,
@@ -240,18 +205,15 @@ impl ReconcilerError for Error {
240205
Error::BuildRbacResources { .. } => None,
241206
Error::FailedToCreatePdb { .. } => None,
242207
Error::GetRequiredLabels { .. } => None,
243-
Error::FailedToValidateAuthenticationMethod { .. } => None,
244208
Error::InvalidKafkaCluster { .. } => None,
245209
Error::MisconfiguredKafkaCluster { .. } => None,
246-
Error::ResolveProductImage { .. } => None,
247210
Error::ParseRole { .. } => None,
248211
Error::BuildStatefulset { .. } => None,
249212
Error::BuildConfigMap { .. } => None,
250213
Error::BuildService { .. } => None,
251214
Error::BuildListener { .. } => None,
252215
Error::InvalidKafkaListeners { .. } => None,
253216
Error::BuildPodDescriptors { .. } => None,
254-
Error::GetOpaConfig { .. } => None,
255217
}
256218
}
257219
}
@@ -270,15 +232,28 @@ pub async fn reconcile_kafka(
270232

271233
let client = &ctx.client;
272234

273-
let resolved_product_image = kafka
274-
.spec
275-
.image
276-
.resolve(
277-
CONTAINER_IMAGE_BASE_NAME,
278-
&ctx.operator_environment.image_repository,
279-
crate::built_info::PKG_VERSION,
280-
)
281-
.context(ResolveProductImageSnafu)?;
235+
// dereference (client required)
236+
let dereferenced_objects = dereference::dereference(client, kafka)
237+
.await
238+
.context(DereferenceSnafu)?;
239+
240+
// validate (no client required)
241+
let validate::ValidatedInputs {
242+
resolved_product_image,
243+
kafka_security,
244+
validated_role_config: validated_config,
245+
} = validate::validate(
246+
kafka,
247+
&dereferenced_objects,
248+
&ctx.operator_environment,
249+
&ctx.product_config,
250+
)
251+
.context(ValidateClusterSnafu)?;
252+
253+
let opa_connect = dereferenced_objects
254+
.resolved_authorization_config
255+
.as_ref()
256+
.map(|auth_config| auth_config.opa_connect.clone());
282257

283258
let mut cluster_resources = ClusterResources::new(
284259
APP_NAME,
@@ -290,37 +265,6 @@ pub async fn reconcile_kafka(
290265
)
291266
.context(CreateClusterResourcesSnafu)?;
292267

293-
let validated_config = validated_product_config(
294-
kafka,
295-
&resolved_product_image.product_version,
296-
&ctx.product_config,
297-
)?;
298-
299-
// Assemble the OPA connection string from the discovery and the given path if provided
300-
// Will be passed as --override parameter in the cli in the stateful set
301-
let opa_config = &kafka
302-
.spec
303-
.cluster_config
304-
.authorization
305-
.clone()
306-
.get_opa_config(client, kafka)
307-
.await
308-
.context(GetOpaConfigSnafu)?;
309-
310-
let opa_connect = opa_config
311-
.as_ref()
312-
.map(|auth_config| auth_config.opa_connect.clone());
313-
314-
let opa_secret_class = if let Some(opa_config) = opa_config.as_ref() {
315-
opa_config.secret_class.to_owned()
316-
} else {
317-
None
318-
};
319-
320-
let kafka_security = KafkaTlsSecurity::new_from_kafka_cluster(client, kafka, opa_secret_class)
321-
.await
322-
.context(FailedToInitializeSecurityContextSnafu)?;
323-
324268
tracing::debug!(
325269
kerberos_enabled = kafka_security.has_kerberos_enabled(),
326270
kerberos_secret_class = ?kafka_security.kerberos_secret_class(),
@@ -329,10 +273,6 @@ pub async fn reconcile_kafka(
329273
"The following security settings are used"
330274
);
331275

332-
kafka_security
333-
.validate_authentication_methods()
334-
.context(FailedToValidateAuthenticationMethodSnafu)?;
335-
336276
let mut ss_cond_builder = StatefulSetConditionBuilder::default();
337277

338278
let (rbac_sa, rbac_rolebinding) = build_rbac_resources(
@@ -537,80 +477,3 @@ pub fn error_policy(
537477
_ => Action::requeue(*Duration::from_secs(5)),
538478
}
539479
}
540-
541-
/// Defines all required roles and their required configuration.
542-
///
543-
/// The roles and their configs are then validated and complemented by the product config.
544-
///
545-
/// # Arguments
546-
/// * `resource` - The TrinoCluster containing the role definitions.
547-
/// * `version` - The TrinoCluster version.
548-
/// * `product_config` - The product config to validate and complement the user config.
549-
///
550-
fn validated_product_config(
551-
kafka: &v1alpha1::KafkaCluster,
552-
product_version: &str,
553-
product_config: &ProductConfigManager,
554-
) -> Result<ValidatedRoleConfigByPropertyKind, Error> {
555-
let mut role_config = HashMap::new();
556-
557-
let broker_role = [(
558-
KafkaRole::Broker.to_string(),
559-
(
560-
vec![
561-
PropertyNameKind::File(BROKER_PROPERTIES_FILE.to_string()),
562-
PropertyNameKind::File(JVM_SECURITY_PROPERTIES_FILE.to_string()),
563-
PropertyNameKind::Env,
564-
],
565-
kafka
566-
.broker_role()
567-
.cloned()
568-
.context(MissingKafkaRoleSnafu {
569-
role: KafkaRole::Broker,
570-
})?
571-
.erase(),
572-
),
573-
)]
574-
.into();
575-
576-
let broker_role_config =
577-
transform_all_roles_to_config(kafka, &broker_role).context(GenerateProductConfigSnafu)?;
578-
579-
role_config.extend(broker_role_config);
580-
581-
// TODO: need this if because controller_role() raises an error
582-
if kafka.spec.controllers.is_some() {
583-
let controller_role = [(
584-
KafkaRole::Controller.to_string(),
585-
(
586-
vec![
587-
PropertyNameKind::File(CONTROLLER_PROPERTIES_FILE.to_string()),
588-
PropertyNameKind::File(JVM_SECURITY_PROPERTIES_FILE.to_string()),
589-
PropertyNameKind::Env,
590-
],
591-
kafka
592-
.controller_role()
593-
.cloned()
594-
.context(MissingKafkaRoleSnafu {
595-
role: KafkaRole::Controller,
596-
})?
597-
.erase(),
598-
),
599-
)]
600-
.into();
601-
602-
let controller_role_config = transform_all_roles_to_config(kafka, &controller_role)
603-
.context(GenerateProductConfigSnafu)?;
604-
605-
role_config.extend(controller_role_config);
606-
}
607-
608-
validate_all_roles_and_groups_config(
609-
product_version,
610-
&role_config,
611-
product_config,
612-
false,
613-
false,
614-
)
615-
.context(InvalidProductConfigSnafu)
616-
}

0 commit comments

Comments
 (0)