Skip to content

Commit ef12f34

Browse files
committed
refactor: move ValidatedCluster to controller/mod.rs; add ConfigFileName enum; remove raw KafkaCluster from build_configmap
1 parent 63d820c commit ef12f34

18 files changed

Lines changed: 262 additions & 179 deletions

File tree

rust/operator-binary/src/config/command.rs

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -7,20 +7,24 @@ use stackable_operator::{
77
};
88

99
use crate::crd::{
10-
BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, LOG4J_CONFIG_FILE, LOG4J2_CONFIG_FILE,
11-
STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR,
12-
STACKABLE_LOG_DIR,
13-
role::{broker::BROKER_PROPERTIES_FILE, controller::CONTROLLER_PROPERTIES_FILE},
10+
BROKER_ID_POD_MAP_DIR, ConfigFileName, KafkaPodDescriptor, STACKABLE_CONFIG_DIR,
11+
STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR,
1412
security::KafkaTlsSecurity,
1513
};
1614

1715
/// The JVM options selecting the Kafka log4j/log4j2 config file. Kafka 3.x uses log4j,
1816
/// Kafka 4.0 and higher use log4j2.
1917
pub fn kafka_log_opts(product_version: &str) -> String {
2018
if product_version.starts_with("3.") {
21-
format!("-Dlog4j.configuration=file:{STACKABLE_LOG_CONFIG_DIR}/{LOG4J_CONFIG_FILE}")
19+
format!(
20+
"-Dlog4j.configuration=file:{STACKABLE_LOG_CONFIG_DIR}/{log4j}",
21+
log4j = ConfigFileName::Log4j
22+
)
2223
} else {
23-
format!("-Dlog4j2.configurationFile=file:{STACKABLE_LOG_CONFIG_DIR}/{LOG4J2_CONFIG_FILE}")
24+
format!(
25+
"-Dlog4j2.configurationFile=file:{STACKABLE_LOG_CONFIG_DIR}/{log4j2}",
26+
log4j2 = ConfigFileName::Log4j2
27+
)
2428
}
2529
}
2630

@@ -77,12 +81,13 @@ fn broker_start_command(
7781
cp {config_dir}/{properties_file} /tmp/{properties_file}
7882
config-utils template /tmp/{properties_file}
7983
80-
cp {config_dir}/jaas.properties /tmp/jaas.properties
81-
config-utils template /tmp/jaas.properties
84+
cp {config_dir}/{jaas_file} /tmp/{jaas_file}
85+
config-utils template /tmp/{jaas_file}
8286
",
8387
broker_id_pod_map_dir = BROKER_ID_POD_MAP_DIR,
8488
config_dir = STACKABLE_CONFIG_DIR,
85-
properties_file = BROKER_PROPERTIES_FILE,
89+
properties_file = ConfigFileName::BrokerProperties,
90+
jaas_file = ConfigFileName::Jaas,
8691
};
8792

8893
if kraft_mode {
@@ -92,15 +97,15 @@ fn broker_start_command(
9297
bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command}
9398
bin/kafka-server-start.sh /tmp/{properties_file} &
9499
",
95-
properties_file = BROKER_PROPERTIES_FILE,
100+
properties_file = ConfigFileName::BrokerProperties,
96101
initial_controller_command = initial_controllers_command(&controller_descriptors, product_version),
97102
}
98103
} else {
99104
formatdoc! {"
100105
{common_command}
101106
102107
bin/kafka-server-start.sh /tmp/{properties_file} &",
103-
properties_file = BROKER_PROPERTIES_FILE,
108+
properties_file = ConfigFileName::BrokerProperties,
104109
}
105110
}
106111
}
@@ -172,7 +177,7 @@ pub fn controller_kafka_container_command(
172177
",
173178
remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
174179
config_dir = STACKABLE_CONFIG_DIR,
175-
properties_file = CONTROLLER_PROPERTIES_FILE,
180+
properties_file = ConfigFileName::ControllerProperties,
176181
initial_controller_command = initial_controllers_command(&controller_descriptors, product_version),
177182
create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
178183
}

rust/operator-binary/src/config/jvm.rs

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,7 @@ use stackable_operator::{
66
schemars::JsonSchema,
77
};
88

9-
use crate::crd::{
10-
JVM_SECURITY_PROPERTIES_FILE, METRICS_PORT, STACKABLE_CONFIG_DIR, role::AnyConfig,
11-
};
9+
use crate::crd::{ConfigFileName, METRICS_PORT, STACKABLE_CONFIG_DIR, role::AnyConfig};
1210

1311
const JAVA_HEAP_FACTOR: f32 = 0.8;
1412

@@ -54,7 +52,10 @@ where
5452
// Heap settings
5553
format!("-Xmx{java_heap}"),
5654
format!("-Xms{java_heap}"),
57-
format!("-Djava.security.properties={STACKABLE_CONFIG_DIR}/{JVM_SECURITY_PROPERTIES_FILE}"),
55+
format!(
56+
"-Djava.security.properties={STACKABLE_CONFIG_DIR}/{security}",
57+
security = ConfigFileName::Security
58+
),
5859
format!(
5960
"-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/server.yaml"
6061
),

rust/operator-binary/src/controller/build/config_map.rs

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -9,16 +9,17 @@ use stackable_operator::{
99

1010
use crate::{
1111
controller::{
12-
KAFKA_CONTROLLER_NAME, ValidatedCluster, ValidatedRoleGroupConfig,
13-
build::properties::logging::role_group_config_map_data, build_recommended_labels,
12+
ValidatedCluster, ValidatedRoleGroupConfig,
13+
build::properties::logging::role_group_config_map_data,
1414
},
1515
crd::{
16-
JVM_SECURITY_PROPERTIES_FILE, MetadataManager, STACKABLE_LISTENER_BOOTSTRAP_DIR,
16+
ConfigFileName, MetadataManager, STACKABLE_LISTENER_BOOTSTRAP_DIR,
1717
STACKABLE_LISTENER_BROKER_DIR,
1818
listener::{KafkaListenerConfig, node_address_cmd},
1919
role::AnyConfig,
2020
v1alpha1,
2121
},
22+
kafka_controller::{KAFKA_CONTROLLER_NAME, build_recommended_labels},
2223
};
2324

2425
#[derive(Snafu, Debug)]
@@ -29,10 +30,7 @@ pub enum Error {
2930
rolegroup: RoleGroupRef<v1alpha1::KafkaCluster>,
3031
},
3132

32-
#[snafu(display(
33-
"failed to serialize [{JVM_SECURITY_PROPERTIES_FILE}] for {}",
34-
rolegroup
35-
))]
33+
#[snafu(display("failed to serialize [{}] for {rolegroup}", ConfigFileName::Security))]
3634
JvmSecurityProperties {
3735
source: PropertiesWriterError,
3836
rolegroup: String,
@@ -60,15 +58,14 @@ pub enum Error {
6058

6159
/// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator
6260
pub fn build_rolegroup_config_map(
63-
kafka: &v1alpha1::KafkaCluster,
6461
validated_cluster: &ValidatedCluster,
6562
rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
6663
validated_rg: &ValidatedRoleGroupConfig,
6764
listener_config: &KafkaListenerConfig,
6865
) -> Result<ConfigMap, Error> {
6966
let kafka_security = &validated_cluster.cluster_config.kafka_security;
7067
let resolved_product_image = &validated_cluster.image;
71-
let kafka_config_file_name = validated_rg.config.config_file_name();
68+
let kafka_config_file_name = validated_rg.config.config_file_name().to_string();
7269
let config_overrides = validated_rg
7370
.config_overrides
7471
.config_file_overrides()
@@ -94,11 +91,9 @@ pub fn build_rolegroup_config_map(
9491
&validated_cluster.cluster_config.pod_descriptors,
9592
opa_connect.as_deref(),
9693
kraft_mode,
97-
kafka
98-
.spec
94+
validated_cluster
9995
.cluster_config
100-
.broker_id_pod_config_map_name
101-
.is_some(),
96+
.disable_broker_id_generation,
10297
config_overrides,
10398
),
10499
AnyConfig::Controller(_) => {
@@ -144,15 +139,15 @@ pub fn build_rolegroup_config_map(
144139
})?,
145140
)
146141
.add_data(
147-
JVM_SECURITY_PROPERTIES_FILE,
142+
ConfigFileName::Security.to_string(),
148143
to_java_properties_string(jvm_sec_props.iter()).with_context(|_| {
149144
JvmSecurityPropertiesSnafu {
150145
rolegroup: rolegroup.role_group.clone(),
151146
}
152147
})?,
153148
)
154149
.add_data(
155-
"client.properties",
150+
ConfigFileName::Client.to_string(),
156151
to_java_properties_string(
157152
kafka_security
158153
.client_properties()
@@ -168,7 +163,7 @@ pub fn build_rolegroup_config_map(
168163
// It is processed by `config-utils` to substitute "env:" and "file:" variables
169164
// and this tool currently doesn't support the JAAS login configuration format.
170165
.add_data(
171-
"jaas.properties",
166+
ConfigFileName::Jaas.to_string(),
172167
jaas_config_file(kafka_security.has_kerberos_enabled()),
173168
);
174169

rust/operator-binary/src/controller/build/discovery.rs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,9 @@ use stackable_operator::{
99
};
1010

1111
use crate::{
12-
controller::{KAFKA_CONTROLLER_NAME, ValidatedCluster, build_recommended_labels},
12+
controller::ValidatedCluster,
1313
crd::{role::KafkaRole, v1alpha1},
14+
kafka_controller::{KAFKA_CONTROLLER_NAME, build_recommended_labels},
1415
};
1516

1617
#[derive(Snafu, Debug)]

rust/operator-binary/src/controller/build/properties/logging.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,12 +13,12 @@ use stackable_operator::{
1313
};
1414

1515
use crate::{
16-
controller::MAX_KAFKA_LOG_FILES_SIZE,
1716
crd::{
18-
LOG4J_CONFIG_FILE, LOG4J2_CONFIG_FILE, STACKABLE_LOG_DIR,
17+
ConfigFileName, STACKABLE_LOG_DIR,
1918
role::{AnyConfig, broker::BrokerContainer, controller::ControllerContainer},
2019
v1alpha1,
2120
},
21+
kafka_controller::MAX_KAFKA_LOG_FILES_SIZE,
2222
};
2323

2424
const KAFKA_LOG4J_FILE: &str = "kafka.log4j.xml";
@@ -44,7 +44,7 @@ pub fn role_group_config_map_data(
4444
match product_version.starts_with("3.") {
4545
true => {
4646
configs.insert(
47-
LOG4J_CONFIG_FILE.to_string(),
47+
ConfigFileName::Log4j.to_string(),
4848
log4j_config_if_automatic(
4949
Some(merged_config.kafka_logging()),
5050
container_name,
@@ -55,7 +55,7 @@ pub fn role_group_config_map_data(
5555
}
5656
false => {
5757
configs.insert(
58-
LOG4J2_CONFIG_FILE.to_string(),
58+
ConfigFileName::Log4j2.to_string(),
5959
log4j2_config_if_automatic(
6060
Some(merged_config.kafka_logging()),
6161
container_name,
Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
//! The validated cluster model and the steps that produce it.
2+
//!
3+
//! [`ValidatedCluster`] carries everything the build steps need, resolved once during
4+
//! [`validate`] (after [`dereference`]) so downstream code never re-derives it or
5+
//! touches the raw [`v1alpha1::KafkaCluster`] spec. The reconcile loop that consumes
6+
//! it lives in [`crate::kafka_controller`].
7+
8+
use std::{borrow::Cow, collections::BTreeMap};
9+
10+
use stackable_operator::{
11+
commons::product_image_selection::ResolvedProductImage,
12+
kube::{Resource, api::ObjectMeta},
13+
v2::types::{
14+
kubernetes::{NamespaceName, Uid},
15+
operator::ClusterName,
16+
},
17+
};
18+
19+
pub(crate) mod build;
20+
pub(crate) mod dereference;
21+
pub(crate) mod validate;
22+
23+
use crate::{
24+
crd::{
25+
KafkaPodDescriptor, MetadataManager,
26+
authorization::KafkaAuthorizationConfig,
27+
role::{AnyConfig, AnyConfigOverrides, KafkaRole},
28+
security::KafkaTlsSecurity,
29+
v1alpha1,
30+
},
31+
framework::role_utils::RoleGroupConfig,
32+
};
33+
34+
pub type RoleGroupName = String;
35+
36+
/// The validated cluster. Carries everything the build steps need, resolved once
37+
/// here so downstream code never re-derives it or touches the raw spec.
38+
///
39+
/// The cluster identity (`name`, `namespace`, `uid`) is captured here so that owner
40+
/// references for child objects can be built straight from this struct (via its
41+
/// [`Resource`] impl) without threading the raw [`v1alpha1::KafkaCluster`] around.
42+
/// This mirrors the hive-/opensearch-operator's `ValidatedCluster`.
43+
pub struct ValidatedCluster {
44+
/// `ObjectMeta` carrying `name`, `namespace` and `uid`, so this struct can act as the
45+
/// owner [`Resource`] for child objects.
46+
metadata: ObjectMeta,
47+
pub name: ClusterName,
48+
pub namespace: NamespaceName,
49+
pub image: ResolvedProductImage,
50+
pub cluster_config: ValidatedClusterConfig,
51+
pub role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
52+
}
53+
54+
impl ValidatedCluster {
55+
pub fn new(
56+
name: ClusterName,
57+
namespace: NamespaceName,
58+
uid: Uid,
59+
image: ResolvedProductImage,
60+
cluster_config: ValidatedClusterConfig,
61+
role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
62+
) -> Self {
63+
Self {
64+
metadata: ObjectMeta {
65+
name: Some(name.to_string()),
66+
namespace: Some(namespace.to_string()),
67+
uid: Some(uid.to_string()),
68+
..ObjectMeta::default()
69+
},
70+
name,
71+
namespace,
72+
image,
73+
cluster_config,
74+
role_group_configs,
75+
}
76+
}
77+
}
78+
79+
/// Cluster-wide settings resolved during validation and dereferencing.
80+
///
81+
/// Everything the build steps need is resolved here so they never have to read the
82+
/// raw [`v1alpha1::KafkaCluster`] spec.
83+
pub struct ValidatedClusterConfig {
84+
pub kafka_security: KafkaTlsSecurity,
85+
pub authorization_config: Option<KafkaAuthorizationConfig>,
86+
pub pod_descriptors: Vec<KafkaPodDescriptor>,
87+
pub metadata_manager: MetadataManager,
88+
89+
/// Whether the operator must not generate broker ids itself, because the user
90+
/// supplied a `broker_id_pod_config_map_name`. Resolved from the raw spec during
91+
/// validation so the config-map builder never has to read it.
92+
pub disable_broker_id_generation: bool,
93+
}
94+
95+
/// Lets [`ValidatedCluster`] act as the owner [`Resource`] for child objects, so owner
96+
/// references are built from it (via the captured `metadata`) rather than the raw CR.
97+
impl Resource for ValidatedCluster {
98+
type DynamicType = <v1alpha1::KafkaCluster as Resource>::DynamicType;
99+
type Scope = <v1alpha1::KafkaCluster as Resource>::Scope;
100+
101+
fn kind(dt: &Self::DynamicType) -> Cow<'_, str> {
102+
v1alpha1::KafkaCluster::kind(dt)
103+
}
104+
105+
fn group(dt: &Self::DynamicType) -> Cow<'_, str> {
106+
v1alpha1::KafkaCluster::group(dt)
107+
}
108+
109+
fn version(dt: &Self::DynamicType) -> Cow<'_, str> {
110+
v1alpha1::KafkaCluster::version(dt)
111+
}
112+
113+
fn plural(dt: &Self::DynamicType) -> Cow<'_, str> {
114+
v1alpha1::KafkaCluster::plural(dt)
115+
}
116+
117+
fn meta(&self) -> &ObjectMeta {
118+
&self.metadata
119+
}
120+
121+
fn meta_mut(&mut self) -> &mut ObjectMeta {
122+
&mut self.metadata
123+
}
124+
}
125+
126+
/// A validated, merged Kafka role-group config.
127+
///
128+
/// The merged config fragment is wrapped in [`AnyConfig`] and the merged
129+
/// `configOverrides` in [`AnyConfigOverrides`], so a single role-agnostic type
130+
/// carries both broker and controller role groups (their concrete config and
131+
/// override types differ). Produced via the local-`framework`
132+
/// [`with_validated_config`](crate::framework::role_utils::with_validated_config).
133+
pub type ValidatedRoleGroupConfig = RoleGroupConfig<
134+
AnyConfig,
135+
stackable_operator::role_utils::JavaCommonConfig,
136+
AnyConfigOverrides,
137+
>;

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -191,6 +191,11 @@ pub fn validate(
191191
authorization_config: dereferenced_objects.authorization_config,
192192
pod_descriptors,
193193
metadata_manager,
194+
disable_broker_id_generation: kafka
195+
.spec
196+
.cluster_config
197+
.broker_id_pod_config_map_name
198+
.is_some(),
194199
},
195200
role_group_configs,
196201
))

0 commit comments

Comments
 (0)