Skip to content

Commit 03bcb41

Browse files
maltesandermaltesander
committed
fix todos
1 parent 934f6e1 commit 03bcb41

5 files changed

Lines changed: 43 additions & 41 deletions

File tree

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

Lines changed: 29 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,3 @@
1-
use std::collections::BTreeMap;
2-
31
use indoc::formatdoc;
42
use stackable_operator::{
53
product_logging::framework::{
@@ -18,6 +16,7 @@ use crate::crd::{
1816
KAFKA_LISTENERS, KAFKA_NODE_ID, KafkaRole, broker::BROKER_PROPERTIES_FILE,
1917
controller::CONTROLLER_PROPERTIES_FILE,
2018
},
19+
security::KafkaTlsSecurity,
2120
v1alpha1,
2221
};
2322

@@ -28,7 +27,7 @@ pub fn broker_kafka_container_commands(
2827
controller_descriptors: Vec<KafkaPodDescriptor>,
2928
kafka_listeners: &KafkaListenerConfig,
3029
opa_connect_string: Option<&str>,
31-
kerberos_enabled: bool,
30+
kafka_security: &KafkaTlsSecurity,
3231
) -> String {
3332
formatdoc! {"
3433
{COMMON_BASH_TRAP_FUNCTIONS}
@@ -44,11 +43,11 @@ pub fn broker_kafka_container_commands(
4443
",
4544
remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
4645
create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR),
47-
set_realm_env = match kerberos_enabled {
46+
set_realm_env = match kafka_security.has_kerberos_enabled() {
4847
true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {})", STACKABLE_KERBEROS_KRB5_PATH),
4948
false => "".to_string(),
5049
},
51-
broker_start_command = broker_start_command(kafka, cluster_id, controller_descriptors, kafka_listeners, opa_connect_string, kerberos_enabled),
50+
broker_start_command = broker_start_command(kafka, cluster_id, controller_descriptors, kafka_listeners, opa_connect_string, kafka_security),
5251
}
5352
}
5453

@@ -58,7 +57,7 @@ fn broker_start_command(
5857
controller_descriptors: Vec<KafkaPodDescriptor>,
5958
kafka_listeners: &KafkaListenerConfig,
6059
opa_connect_string: Option<&str>,
61-
kerberos_enabled: bool,
60+
kafka_security: &KafkaTlsSecurity,
6261
) -> String {
6362
let opa_config = match opa_connect_string {
6463
None => "".to_string(),
@@ -67,7 +66,7 @@ fn broker_start_command(
6766
}
6867
};
6968

70-
let jaas_config = match kerberos_enabled {
69+
let jaas_config = match kafka_security.has_kerberos_enabled() {
7170
true => {
7271
let service_name = KafkaRole::Broker.kerberos_service_name();
7372
let broker_address = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR);
@@ -78,6 +77,8 @@ fn broker_start_command(
7877
false => "".to_string(),
7978
};
8079

80+
let client_port = kafka_security.client_port();
81+
8182
// TODO: copy to tmp? mount readwrite folder?
8283
if kafka.is_controller_configured() {
8384
formatdoc! {"
@@ -95,8 +96,8 @@ fn broker_start_command(
9596
",
9697
config_dir = STACKABLE_CONFIG_DIR,
9798
properties_file = BROKER_PROPERTIES_FILE,
98-
bootstrap_servers = to_bootstrap_servers(&controller_descriptors),
99-
initial_controllers = to_initial_controllers(&controller_descriptors),
99+
bootstrap_servers = to_bootstrap_servers(&controller_descriptors, client_port),
100+
initial_controllers = to_initial_controllers(&controller_descriptors, client_port),
100101
listeners = kafka_listeners.listeners(),
101102
advertised_listeners = kafka_listeners.advertised_listeners(),
102103
listener_security_protocol_map = kafka_listeners.listener_security_protocol_map(),
@@ -123,8 +124,9 @@ fn broker_start_command(
123124
pub fn controller_kafka_container_command(
124125
cluster_id: &str,
125126
controller_descriptors: Vec<KafkaPodDescriptor>,
126-
server_start_overrides: BTreeMap<String, String>,
127+
kafka_security: &KafkaTlsSecurity,
127128
) -> String {
129+
let client_port = kafka_security.client_port();
128130
// TODO: copy to tmp? mount readwrite folder?
129131
formatdoc! {"
130132
{COMMON_BASH_TRAP_FUNCTIONS}
@@ -141,57 +143,54 @@ pub fn controller_kafka_container_command(
141143
echo \"{KAFKA_LISTENER_SECURITY_PROTOCOL_MAP}={listener_security_protocol_map}\" >> /tmp/{properties_file}
142144
143145
bin/kafka-storage.sh format --cluster-id {cluster_id} --config /tmp/{properties_file} --initial-controllers {initial_controllers} --ignore-formatted
144-
bin/kafka-server-start.sh /tmp/{properties_file} {overrides} &
146+
bin/kafka-server-start.sh /tmp/{properties_file} &
145147
146148
wait_for_termination $!
147149
{create_vector_shutdown_file_command}
148150
",
149151
remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
150152
config_dir = STACKABLE_CONFIG_DIR,
151153
properties_file = CONTROLLER_PROPERTIES_FILE,
152-
bootstrap_servers = to_bootstrap_servers(&controller_descriptors),
153-
listeners = to_listeners(),
154+
bootstrap_servers = to_bootstrap_servers(&controller_descriptors, client_port),
155+
listeners = to_listeners(client_port),
154156
listener_security_protocol_map = to_listener_security_protocol_map(),
155-
initial_controllers = to_initial_controllers(&controller_descriptors),
156-
overrides = to_kafka_overrides(server_start_overrides),
157+
initial_controllers = to_initial_controllers(&controller_descriptors, client_port),
157158
create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
158159
}
159160
}
160161

161-
fn to_listeners() -> String {
162+
fn to_listeners(port: u16) -> String {
162163
// TODO:
163164
// - document that variables are set in stateful set
164165
// - customize listener (CONTROLLER)
165-
// - customize port
166-
"CONTROLLER://$POD_NAME.$ROLEGROUP_REF.$NAMESPACE.svc.$CLUSTER_DOMAIN:9093".to_string()
166+
format!("CONTROLLER://$POD_NAME.$ROLEGROUP_REF.$NAMESPACE.svc.$CLUSTER_DOMAIN:{port}")
167167
}
168168

169169
fn to_listener_security_protocol_map() -> String {
170170
// TODO: make configurable
171171
"CONTROLLER:PLAINTEXT".to_string()
172172
}
173173

174-
fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor]) -> String {
174+
fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor], port: u16) -> String {
175175
controller_descriptors
176176
.iter()
177-
.map(|desc| desc.as_voter())
177+
.map(|desc| desc.as_voter(port))
178178
.collect::<Vec<String>>()
179179
.join(",")
180180
}
181181

182-
fn to_bootstrap_servers(controller_descriptors: &[KafkaPodDescriptor]) -> String {
182+
fn to_bootstrap_servers(controller_descriptors: &[KafkaPodDescriptor], port: u16) -> String {
183183
controller_descriptors
184184
.iter()
185-
// TODO: make port configureable
186-
.map(|desc| format!("{fqdn}:{port}", fqdn = desc.fqdn(), port = 9093))
185+
.map(|desc| format!("{fqdn}:{port}", fqdn = desc.fqdn()))
187186
.collect::<Vec<String>>()
188187
.join(",")
189188
}
190189

191-
fn to_kafka_overrides(overrides: BTreeMap<String, String>) -> String {
192-
overrides
193-
.iter()
194-
.map(|(key, value)| format!("--override \"{key}={value}\""))
195-
.collect::<Vec<String>>()
196-
.join(" ")
197-
}
190+
// fn to_kafka_overrides(overrides: BTreeMap<String, String>) -> String {
191+
// overrides
192+
// .iter()
193+
// .map(|(key, value)| format!("--override \"{key}={value}\""))
194+
// .collect::<Vec<String>>()
195+
// .join(" ")
196+
// }

‎rust/operator-binary/src/crd/mod.rs‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -259,6 +259,7 @@ impl v1alpha1::KafkaCluster {
259259
})
260260
.collect(),
261261

262+
// TODO: this does not work for multiple rolegroups (the index / replica)
262263
KafkaRole::Controller => self
263264
.controller_role()
264265
.iter()
@@ -269,7 +270,7 @@ impl v1alpha1::KafkaCluster {
269270
.flat_map(move |(rolegroup_name, rolegroup)| {
270271
let rolegroup_ref = self.rolegroup_ref(kafka_role, rolegroup_name);
271272
let ns = ns.clone();
272-
(0..rolegroup.replicas.unwrap_or(0)).map(move |i| KafkaPodDescriptor {
273+
(0..rolegroup.replicas.unwrap_or(0)).map(move |i: u16| KafkaPodDescriptor {
273274
namespace: ns.clone(),
274275
role_group_service_name: rolegroup_ref.object_name(),
275276
replica: i,
@@ -329,13 +330,11 @@ impl KafkaPodDescriptor {
329330
/// * controller-0 is the replica's host,
330331
/// * 1234 is the replica's port.
331332
// TODO(@maltesander): Even though the used Uuid states to be type 4 it does not work... 0000000000-00000000000 works...
332-
pub fn as_voter(&self) -> String {
333+
pub fn as_voter(&self, port: u16) -> String {
333334
format!(
334335
"{replica}@{fqdn}:{port}:0000000000-{replica:0>11}",
335336
replica = self.replica,
336337
fqdn = self.fqdn(),
337-
// TODO: make port configureable
338-
port = 9093
339338
)
340339
}
341340
}

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -389,7 +389,6 @@ pub async fn reconcile_kafka(
389389
.context(BuildStatefulsetSnafu)?,
390390
};
391391

392-
// TODO: broker / controller?
393392
if let AnyConfig::Broker(broker_config) = merged_config {
394393
let rg_bootstrap_listener = build_broker_rolegroup_bootstrap_listener(
395394
kafka,

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

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,11 @@ use stackable_operator::{
1010
role_utils::RoleGroupRef,
1111
};
1212

13-
use crate::crd::{STACKABLE_LOG_DIR, role::AnyConfig, v1alpha1};
13+
use crate::crd::{
14+
STACKABLE_LOG_DIR,
15+
role::{AnyConfig, broker::BrokerContainer, controller::ControllerContainer},
16+
v1alpha1,
17+
};
1418

1519
pub const LOG4J_CONFIG_FILE: &str = "log4j.properties";
1620
pub const KAFKA_LOG_FILE: &str = "kafka.log4j.xml";
@@ -59,8 +63,10 @@ pub fn extend_role_group_config_map(
5963
cm_builder,
6064
Some(merged_config.kafka_logging()),
6165
LOG4J_CONFIG_FILE,
62-
// TODO: configure?
63-
"kafka",
66+
match merged_config {
67+
AnyConfig::Broker(_) => BrokerContainer::Kafka.to_string(),
68+
AnyConfig::Controller(_) => ControllerContainer::Kafka.to_string(),
69+
},
6470
KAFKA_LOG_FILE,
6571
MAX_KAFKA_LOG_FILES_SIZE,
6672
);

‎rust/operator-binary/src/resource/statefulset.rs‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -301,7 +301,7 @@ pub fn build_broker_rolegroup_statefulset(
301301
.context(BuildPodDescriptorsSnafu)?,
302302
&kafka_listeners,
303303
opa_connect_string,
304-
kafka_security.has_kerberos_enabled(),
304+
kafka_security,
305305
)])
306306
.add_env_var(
307307
"EXTRA_ARGS",
@@ -629,8 +629,7 @@ pub fn build_controller_rolegroup_statefulset(
629629
kafka
630630
.pod_descriptors(kafka_role, cluster_info)
631631
.context(BuildPodDescriptorsSnafu)?,
632-
// TODO: fix overrides
633-
BTreeMap::new(),
632+
kafka_security,
634633
)])
635634
.add_env_var(
636635
"EXTRA_ARGS",

0 commit comments

Comments
 (0)