diff --git a/.gitignore b/.gitignore index 696bc411..0ff5d1ec 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,5 @@ tilt_options.json .envrc .DS_Store + +.worktrees/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e3b1a1a..dd3a055b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- KRaft controller replicas can now be scaled up and down on a running cluster: a new + `quorum-manager` sidecar container on each controller pod admits itself into the KRaft + voter set on startup and removes itself before termination ([#NNNN]). + ### Changed - Internal operator refactoring: introduce a build() step in the reconciler that @@ -16,6 +22,9 @@ All notable changes to this project will be documented in this file. - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#998]). - The reconciler now applies resources and derives the cluster status in discrete apply and update_status steps ([#1000]). +- BREAKING: KRaft mode now requires Kafka 3.9.0 or later; Kafka 3.7.x is no longer supported and its previous + special-casing has been removed entirely, rather than narrowed. Running KRaft mode on an unsupported Kafka + version is undefined behavior ([#NNNN]). ### Fixed @@ -28,6 +37,7 @@ All notable changes to this project will be documented in this file. [#994]: https://github.com/stackabletech/kafka-operator/pull/994 [#998]: https://github.com/stackabletech/kafka-operator/pull/998 [#1000]: https://github.com/stackabletech/kafka-operator/pull/1000 +[#NNNN]: https://github.com/stackabletech/kafka-operator/pull/NNNN ## [26.7.0] - 2026-07-21 diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 455188c9..805fcac2 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -16,6 +16,11 @@ WARNING: The Stackable Operator for Apache Kafka currently does not support auto * Full Replacement: Kafka 4.0.0 (2025) removes ZooKeeper completely. * Migration: Tools exist to migrate from ZooKeeper to KRaft, but new deployments should start with KRaft. +IMPORTANT: The Stackable Operator for Apache Kafka requires Kafka 3.9.0 or later for KRaft mode. Kafka 3.7.x is not +supported: the operator relies on the `kafka-storage.sh format --initial-controllers` option and the KIP-853 dynamic +quorum tooling, both of which require 3.9.0+. Running KRaft mode on an unsupported Kafka version is undefined +behavior, up to and including `kafka-storage.sh` rejecting the operator-generated formatting command outright. + == Configuration The Stackable Kafka operator introduces a new xref:concepts:roles-and-role-groups.adoc[role] in the KafkaCluster CRD called KRaft `Controller`. @@ -85,13 +90,27 @@ KRaft mode requires major configuration changes compared to ZooKeeper: * `cluster-id`: This is set to the `metadata.name` of the KafkaCluster resource during initial formatting * `node.id`: This is a calculated integer, hashed from the `role` and `rolegroup` and added `replica` id. * `process.roles`: Will always only be `broker` or `controller`. Mixed `broker,controller` servers are not supported. -* The operator configures a static voter list containing the controller pods. Controllers are not dynamically managed. +* When Kerberos is not enabled, each controller pod runs an additional `quorum-manager` sidecar container (requires + Kafka 3.9.0 or later, see the minimum-version note above). On startup it admits the pod into the KRaft voter set + (`kafka-metadata-quorum.sh add-controller`), and on pod termination (`preStop`) it removes the pod from the voter + set again (`remove-controller`), but only if doing so would not remove the last remaining voter. == Known Issues * Automatic migration from Apache ZooKeeper to KRaft is not supported. -* Scaling controller replicas might lead to unstable clusters. * Kerberos is currently not supported for KRaft in all versions. +* Scaling controllers down to a single replica is not verified under the sidecar-based mechanism described above; + only scale-downs that keep an odd number of controllers greater than one have been tested. +* If a `remove-controller` call fails or times out during pod termination (for example, no reachable leader within + the pod's grace period), the pod terminates anyway and can leave a stale voter entry in the quorum behind. This is + not fully automatic in every case and may require manual cleanup with `kafka-metadata-quorum.sh remove-controller`. +* When a new controller is added on scale-up, it is formatted with `kafka-storage.sh format --initial-controllers` + listing the *full* desired voter set, including itself, rather than following Kafka's documented procedure for + joining an already-formed quorum (format with no initial controllers, then explicitly `add-controller`). In manual + live testing this has been observed to work correctly: the new pod starts as an `observer` and the sidecar's + `add-controller` call succeeds. However, this has not yet been confirmed end-to-end through the fully automated + sidecar path on a live cluster. This is a known, tracked risk to verify before or soon after this feature ships; + changing the format-step behavior itself would be a separate, larger design change. == Troubleshooting @@ -108,10 +127,19 @@ Likely caused by controller resource starvation or unstable Kubernetes schedulin Ensure Kafka version 3.9.x and higher and follow the official migration documentation. The Stackable Kafka operator currently does not support the migration. -=== Scaling issues +=== Scaling controllers + +Controller replicas can be scaled up and down on a running cluster. A per-pod `quorum-manager` sidecar admits and +removes the pod from the KRaft voter set as described under "Internal operator details" above. This requires Kafka +3.9.0 or later (see the minimum-version note in "Overview"), which supports the +https://developers.redhat.com/articles/2024/11/27/dynamic-kafka-controller-quorum[KIP-853 dynamic quorum tooling]. + +Scaling more than one controller down at a time is processed one pod at a time (`OrderedReady` pod management), not +in parallel, so that each pod's removal from the voter set can complete before the next one is terminated. -The https://developers.redhat.com/articles/2024/11/27/dynamic-kafka-controller-quorum[Dynamic scaling] is only supported from Kafka version 3.9.0. -If you are using older versions, automatic scaling may not work properly (e.g. adding or removing controller replicas). +This has been validated on a live cluster and is covered by unit tests, but a full end-to-end scale-up/scale-down +kuttl test suite run has not yet produced a clean pass; treat controller scaling as functional but not yet fully +verified end-to-end. == Kraft migration guide diff --git a/docs/superpowers/plans/2026-08-14-kraft-dynamic-voter-membership.md b/docs/superpowers/plans/2026-08-14-kraft-dynamic-voter-membership.md new file mode 100644 index 00000000..723f7612 --- /dev/null +++ b/docs/superpowers/plans/2026-08-14-kraft-dynamic-voter-membership.md @@ -0,0 +1,1008 @@ +# KRaft Dynamic Voter Membership Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let KRaft controller replicas be scaled up and down on a running Kafka cluster, by having each controller pod manage its own quorum membership via a sidecar container instead of the operator talking to the live cluster. + +**Architecture:** A new `quorum-manager` sidecar container (reusing the Kafka product image) runs alongside the `kafka` container on controller pods only. Its main process loops, admitting itself as a voter (`kafka-metadata-quorum.sh add-controller`) while its local Raft state is `observer`. Its `preStop` hook checks a majority-safety condition and removes itself (`remove-controller`) before termination. The controller StatefulSet switches `podManagementPolicy` to `OrderedReady` so Kubernetes drains controllers one at a time on scale-down. All of this is gated to Kafka versions that support KIP-853 dynamic quorum tooling (everything except the `3.7.x` line, mirroring the existing `--initial-controllers` version check). + +**Tech Stack:** Rust (`stackable-operator`, `kube-rs` builder types), Bash (sidecar scripts), Kafka's `kafka-metadata-quorum.sh` CLI, kuttl (integration tests). + +**Spec:** `docs/superpowers/specs/2026-08-14-kraft-dynamic-voter-membership-design.md` + +## Global Constraints + +- Sidecar is added only for the controller role (`build_controller_rolegroup_statefulset`), never brokers. +- Sidecar is added only when `!resolved_product_image.product_version.starts_with("3.7")` — same literal-prefix check style as `initial_controllers_command` (`rust/operator-binary/src/controller/build/command.rs:198-211`) and `uses_legacy_log4j` (`rust/operator-binary/src/controller/build/properties/mod.rs:55-57`). +- Sidecar is not added when `kafka_security.has_kerberos_enabled()` is true — Kerberos for KRaft is already a documented unsupported combination (`docs/modules/kafka/pages/usage-guide/kraft-controller.adoc`, Known Issues), and the sidecar's admin-client properties file only handles the TLS/SSL case. +- The sidecar's `preStop` script must always exit `0`, regardless of whether `remove-controller` succeeded — it must never block pod termination. +- The sidecar targets the quorum's bootstrap servers (its peers), never `localhost` for the admin-client calls — its own `kafka` container may be concurrently shutting down. +- No new Kubernetes RBAC, no new reconcile phase, no CRD status field. All Rust changes are confined to the `build` phase. +- Every `rust/` change must pass `cargo build`, `cargo test`, `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check` before being considered done, per this repo's existing verification gate. CRDs/docs are regenerated (`make regenerate-charts`) whenever the CRD schema changes. +- A CHANGELOG entry is added in the same commit as the change it documents, under a new `### Added` section (there is currently no `### Added` section under `## [Unreleased]` in `CHANGELOG.md`). + +--- + +## Task 1: Version gate helper — `supports_dynamic_quorum` + +**Files:** + +- Modify: `rust/operator-binary/src/controller/build/properties/mod.rs` +- Test: same file, `#[cfg(test)] mod tests` (create if absent — check first, this file may already have one) + +**Interfaces:** + +- Produces: `pub fn supports_dynamic_quorum(product_version: &str) -> bool` — used by Task 4 (sidecar container gating) and Task 7 (kuttl test gating verification). + +- [ ] **Step 1: Check for an existing test module in this file** + +Run: `grep -n "mod tests" rust/operator-binary/src/controller/build/properties/mod.rs` + +If it exists, note the line number — new tests go inside it. If not, one will be created in Step 3. + +- [ ] **Step 2: Write the failing test** + +Add near the existing `uses_legacy_log4j` function (`rust/operator-binary/src/controller/build/properties/mod.rs:55-57`): + +```rust +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn dynamic_quorum_is_supported_from_3_9_onwards() { + assert!(supports_dynamic_quorum("3.9.2")); + assert!(supports_dynamic_quorum("4.1.1")); + assert!(supports_dynamic_quorum("4.2.1")); + } + + #[test] + fn dynamic_quorum_is_not_supported_on_3_7() { + assert!(!supports_dynamic_quorum("3.7.2")); + } +} +``` + +If a `mod tests` block already exists in this file, add these two `#[test]` functions inside it instead of writing a new module, and skip the `use super::*;` line if it's already present. + +- [ ] **Step 3: Run the test to verify it fails** + +Run: `cargo test -p stackable-kafka-operator-binary supports_dynamic_quorum 2>&1 | tail -30` +Expected: compile error, `supports_dynamic_quorum` not found. + +- [ ] **Step 4: Write the minimal implementation** + +Add next to `uses_legacy_log4j`: + +```rust +/// Whether this Kafka version supports the KIP-853 dynamic KRaft quorum tooling +/// (`kafka-metadata-quorum.sh add-controller` / `remove-controller`) needed to change +/// the voter set of an already-formed quorum. Mirrors the existing 3.7.x carve-out +/// already used for `--initial-controllers` (see `initial_controllers_command` in +/// `build/command.rs`). +pub fn supports_dynamic_quorum(product_version: &str) -> bool { + !product_version.starts_with("3.7") +} +``` + +- [ ] **Step 5: Run the test to verify it passes** + +Run: `cargo test -p stackable-kafka-operator-binary supports_dynamic_quorum 2>&1 | tail -30` +Expected: both tests PASS. + +- [ ] **Step 6: Commit** + +```bash +git add rust/operator-binary/src/controller/build/properties/mod.rs +git commit -m "feat: add supports_dynamic_quorum version gate + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 2: Admin-client properties file for the sidecar's TLS config + +**Files:** + +- Modify: `rust/operator-binary/src/controller/build/security.rs` +- Modify: `rust/operator-binary/src/controller/build/resource/config_map.rs` +- Test: `rust/operator-binary/src/controller/build/security.rs`, existing `#[cfg(test)] mod tests` (lines 653-992) + +**Interfaces:** + +- Consumes: `push_client_ssl_stores` (`security.rs:374-388`), `push_client_ssl_truststore` (`security.rs:392-405`), `STACKABLE_TLS_KAFKA_INTERNAL_DIR` (`security.rs:44`), `PROPERTY_SECURITY_PROTOCOL` (`security.rs:41`), `ValidatedKafkaSecurity` (already used throughout this file). +- Produces: `pub fn controller_admin_client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Option)>` — consumed by Task 3's ConfigMap wiring and referenced by path (`/stackable/config/admin-client.properties`) in Task 4's sidecar scripts. + +The existing `client.properties` (built by `client_properties()`, `security.rs:165-222`) is unusable for the sidecar: it points at `/stackable/tls-kafka-server`, a directory that is only mounted on broker pods (`add_broker_volume_and_volume_mounts`), never on controller pods. Controller pods only mount `/stackable/tls-kafka-internal` (`add_controller_volume_and_volume_mounts`, `security.rs:301-337`). This task adds a new properties builder pointed at that directory instead, using unprefixed `security.protocol`/`ssl.*` keys (the ones a plain Kafka admin client / `--command-config` needs), as opposed to the `listener.name.controller.ssl.*`-prefixed keys `controller_config_settings()` writes for the broker/controller's own server-side listener config. + +- [ ] **Step 1: Write the failing test** + +Add inside the existing `#[cfg(test)] mod tests` block in `security.rs` (near the other `*_properties`-style tests — check the existing fixtures `plaintext()`, `server_tls()`, `client_auth_tls()`, `as_map()` around lines 653-992 and reuse them): + +```rust + #[test] + fn controller_admin_client_properties_uses_the_internal_tls_directory() { + let security = server_tls(); + let props = as_map(controller_admin_client_properties(&security)); + + assert_eq!(props.get("security.protocol"), Some(&"SSL".to_string())); + assert_eq!( + props.get("ssl.truststore.location"), + Some(&"/stackable/tls-kafka-internal/truststore.p12".to_string()) + ); + assert_eq!(props.get("ssl.truststore.type"), Some(&"PKCS12".to_string())); + } + + #[test] + fn controller_admin_client_properties_includes_keystore_when_client_auth_is_required() { + let security = client_auth_tls(); + let props = as_map(controller_admin_client_properties(&security)); + + assert_eq!( + props.get("ssl.keystore.location"), + Some(&"/stackable/tls-kafka-internal/keystore.p12".to_string()) + ); + } + + #[test] + fn controller_admin_client_properties_is_plaintext_when_no_tls_is_configured() { + let security = plaintext(); + let props = as_map(controller_admin_client_properties(&security)); + + assert_eq!(props.get("security.protocol"), None); + assert_eq!(props.get("ssl.truststore.location"), None); + } +``` + +Check the exact names/signatures of `server_tls()`, `client_auth_tls()`, `plaintext()`, and `as_map()` in the existing test module before using them — copy their exact fixture-building style if these names differ slightly. + +- [ ] **Step 2: Run the tests to verify they fail** + +Run: `cargo test -p stackable-kafka-operator-binary controller_admin_client_properties 2>&1 | tail -40` +Expected: compile error, `controller_admin_client_properties` not found. + +- [ ] **Step 3: Write the minimal implementation** + +Add to `security.rs`, near `client_properties()` (around line 165), following the same shape (`Vec<(String, Option)>` of key/value pairs, `None` values filtered out by the caller): + +```rust +/// Client-side (unprefixed `security.protocol`/`ssl.*`) properties for an admin CLI tool +/// (e.g. `kafka-metadata-quorum.sh`) talking to the CONTROLLER listener from *inside* a +/// controller pod, over the `tls-kafka-internal` volume mounted by +/// `add_controller_volume_and_volume_mounts`. +/// +/// This is deliberately separate from `client_properties()`: that function points at +/// `/stackable/tls-kafka-server`, a directory that is only mounted on broker pods. +pub fn controller_admin_client_properties( + security: &ValidatedKafkaSecurity, +) -> Vec<(String, Option)> { + let mut properties = vec![]; + + if security.tls_internal_secret_class().is_some() { + properties.push(( + PROPERTY_SECURITY_PROTOCOL.to_string(), + Some("SSL".to_string()), + )); + push_client_ssl_truststore(&mut properties, STACKABLE_TLS_KAFKA_INTERNAL_DIR); + if security.tls_client_authentication_class().is_some() { + push_client_ssl_stores(&mut properties, STACKABLE_TLS_KAFKA_INTERNAL_DIR); + } + } + + properties +} +``` + +Check the exact method name for "is internal TLS configured" (`tls_internal_secret_class()` is a guess based on the sibling `tls_server_secret_class()`/`tls_client_authentication_class()` naming seen in `kcat_prober_container_commands`, `security.rs:95-161`) — grep for the real accessor: + +Run: `grep -n "fn tls_.*secret_class\|fn tls_client_authentication_class" rust/operator-binary/src/crd/security.rs rust/operator-binary/src/controller/security.rs 2>/dev/null` + +Adjust the method name used above to match what actually exists on `ValidatedKafkaSecurity`. + +- [ ] **Step 4: Run the tests to verify they pass** + +Run: `cargo test -p stackable-kafka-operator-binary controller_admin_client_properties 2>&1 | tail -40` +Expected: all three PASS. + +- [ ] **Step 5: Wire the new properties into the controller rolegroup ConfigMap** + +Read `rust/operator-binary/src/controller/build/resource/config_map.rs:150-175` first to see exactly how `client.properties` is added, then add a sibling entry for the controller role only. Find the `ConfigFileName` enum (grep `enum ConfigFileName`) and add a variant: + +Run: `grep -n "enum ConfigFileName" -A 10 rust/operator-binary/src/controller/build/resource/config_map.rs rust/operator-binary/src/crd/mod.rs 2>/dev/null` + +Add a variant named `AdminClient` (kebab-case via the same derive macros the enum already uses) that serializes to `admin-client.properties`, then add, guarded to the controller role group's `add_data` block (mirroring the `client.properties` call at `config_map.rs:155-165`): + +```rust + .add_data( + ConfigFileName::AdminClient.to_string(), + to_java_properties_string( + controller_admin_client_properties(kafka_security) + .iter() + .filter_map(|(k, v)| v.as_ref().map(|v| (k, v))), + ) + .context(SerializePropertiesSnafu)?, + ) +``` + +Place this call only in the function that builds the **controller** rolegroup ConfigMap, not the broker one — check the function name/boundary by reading the file's structure first (`grep -n "^pub fn\|^fn" rust/operator-binary/src/controller/build/resource/config_map.rs`). + +- [ ] **Step 6: Run the full properties/config_map test suite** + +Run: `cargo test -p stackable-kafka-operator-binary --lib config_map security 2>&1 | tail -60` +Expected: all PASS, no regressions in existing `client.properties`/`controller.properties` tests. + +- [ ] **Step 7: Commit** + +```bash +git add rust/operator-binary/src/controller/build/security.rs rust/operator-binary/src/controller/build/resource/config_map.rs +git commit -m "feat: add admin-client.properties for controller-pod CLI tools + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 3: Expose bootstrap servers and node id to the sidecar + +**Files:** + +- Modify: `rust/operator-binary/src/controller/build/resource/statefulset.rs` + +**Interfaces:** + +- Consumes: `kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Vec` (`rust/operator-binary/src/controller/build/properties/mod.rs:59-71`, already `pub(crate)`), `validated_cluster.pod_descriptors(Some(kafka_role))` (already used at `statefulset.rs:275`, `:505`), `KAFKA_NODE_ID_OFFSET` env var name and `node_id_hash32_offset(...)` (already used at `statefulset.rs:706-709` on the `kafka` container — read that exact block before duplicating it). +- Produces: two env vars the sidecar container (Task 4) will read: `KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS` (comma-joined `host:port` list) and the same `REPLICA_ID`-deriving inputs (`POD_NAME` via downward API, `NODE_ID_OFFSET`) already present on the `kafka` container, so the sidecar's script can compute its own replica/node id exactly as `command.rs:169-170` does inside the `kafka` container's entrypoint. + +This task only adds env vars to the (not-yet-created) sidecar container's builder; Task 4 creates that builder. Do this task by extending `build_controller_rolegroup_statefulset` to compute the values once and store them in local variables the Task 4 diff will consume — do not create the sidecar container yet, since that would make this task's diff untestable on its own. Instead, write a small pure helper function now, unit-test it in isolation, and call it from Task 4. + +- [ ] **Step 1: Write the failing test** + +Add near wherever `kraft_controllers` is exported from (`rust/operator-binary/src/controller/build/properties/mod.rs`), or create a new test in `statefulset.rs` if a test module doesn't exist yet there (check first: `grep -n "mod tests" rust/operator-binary/src/controller/build/resource/statefulset.rs`; if absent, this task creates the module, which Task 6 will also extend): + +```rust +#[cfg(test)] +mod tests { + use crate::controller::build::properties::kraft_controllers; + use crate::crd::mod::KafkaPodDescriptor; // adjust path once the real module path is confirmed + + #[test] + fn quorum_manager_bootstrap_servers_env_value_is_comma_joined_host_ports() { + // Build two minimal KafkaPodDescriptor values for controllers and assert + // kraft_controllers(...).join(",") produces "host1:9093,host2:9093". + // Fill in with the real KafkaPodDescriptor construction used in + // crd/mod.rs's own tests, since its fields are crate-private (pub(crate)). + } +} +``` + +Before writing this test for real, run: + +Run: `grep -n "KafkaPodDescriptor {" rust/operator-binary/src/crd/mod.rs` + +to find an existing test or construction site building a `KafkaPodDescriptor` by hand (its fields are `pub(crate)`, so this must be done from within the `crd` module or via a test already inside `crd/mod.rs`). If no direct constructor is accessible from `statefulset.rs`'s test module, skip a standalone unit test for the joining logic here (it's a one-line `.join(",")` over an already-tested function) and instead verify this wiring via the integration-style test added in Task 6, which builds a full `ValidatedCluster` through the public `validate()` path and inspects the sidecar container's env vars directly. Note that decision in the commit message for this task. + +- [ ] **Step 2: Add the env var to `build_controller_rolegroup_statefulset`** + +In `rust/operator-binary/src/controller/build/resource/statefulset.rs`, inside `build_controller_rolegroup_statefulset` (around line 505, right after the existing `pod_descriptors(Some(kafka_role))` call used for `controller_kafka_container_command`), compute: + +```rust + let controller_pod_descriptors = validated_cluster + .pod_descriptors(Some(kafka_role)) + .context(BuildPodDescriptorsSnafu)?; + let quorum_bootstrap_servers = + crate::controller::build::properties::kraft_controllers(&controller_pod_descriptors) + .join(","); +``` + +Reuse the existing `pod_descriptors(...)` call already present at line 505 rather than calling it twice — read the surrounding code first and thread `controller_pod_descriptors` through to both the existing `controller_kafka_container_command(...)` call and this new binding, instead of calling `pod_descriptors` a second time. + +Store `quorum_bootstrap_servers` in a local variable for Task 4 to consume when building the sidecar container's env vars — do not add it to the `kafka` container's env vars in this task (it's only needed by the sidecar). + +- [ ] **Step 3: Run the build to confirm it still compiles** + +Run: `cargo build -p stackable-kafka-operator-binary 2>&1 | tail -40` +Expected: compiles cleanly. `quorum_bootstrap_servers` will show an "unused variable" warning until Task 4 consumes it — that's expected and acceptable to leave as a `#[allow(unused)]`-free warning between these two tasks only if they're implemented back-to-back in the same session; otherwise prefix with `_` temporarily. Prefer implementing Task 4 immediately after this task in the same sitting so the warning never needs suppressing. + +- [ ] **Step 4: Commit** + +```bash +git add rust/operator-binary/src/controller/build/resource/statefulset.rs +git commit -m "feat: compute quorum bootstrap servers for the controller sidecar + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 4: Build the `quorum-manager` sidecar container + +**Files:** + +- Modify: `rust/operator-binary/src/controller/build/resource/statefulset.rs` +- Modify: `rust/operator-binary/src/controller/build/command.rs` + +**Interfaces:** + +- Consumes: `supports_dynamic_quorum` (Task 1), `controller_admin_client_properties` path convention `/stackable/config/admin-client.properties` (Task 2 — the file this properties struct serializes to, mounted via the existing `STACKABLE_CONFIG_DIR_NAME` volume mount already present on the `kafka` container at `statefulset.rs:293`), `quorum_bootstrap_servers` local variable (Task 3), `METRICS_PORT`/`METRICS_PORT_NAME` (`crd/mod.rs:45-46`), `kafka_security.has_kerberos_enabled()` (already used at `container_ports`, `statefulset.rs:644-668`). +- Produces: the sidecar `Container`, added to the pod via `pod_builder.add_container(...)` — consumed by Task 6's unit tests (which inspect it by container name `"quorum-manager"`) and Task 7's kuttl assertions (which observe its effect on the live cluster). + +- [ ] **Step 1: Add the two script-building functions to `command.rs`** + +Read `rust/operator-binary/src/controller/build/command.rs` in full first (it's 209 lines) to match its existing style (plain `String`/`format!`, no templating engine). Add two new functions near `controller_kafka_container_command`: + +```rust +/// The `kafka-metadata-quorum.sh` binary, referenced by its absolute path (matching every +/// other exec-into-pod usage of a Kafka CLI tool in this repo, e.g. the kuttl test scripts +/// under `tests/templates/kuttl/*/*.sh`), rather than the relative `bin/...` form used only +/// inside the `kafka` container's own entrypoint (which runs with the Kafka install dir as +/// its working directory). +const KAFKA_METADATA_QUORUM_BINARY: &str = "/stackable/kafka/bin/kafka-metadata-quorum.sh"; + +const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/stackable/config/admin-client.properties"; + +/// The sidecar's main-loop command: while this controller's local Raft state is +/// `observer`, repeatedly attempt to admit it into the quorum's voter set. +/// +/// `bootstrap_servers` is the comma-joined `host:port` list produced by +/// `kraft_controllers(...)` (see `build/properties/mod.rs`). +pub fn quorum_manager_container_command(bootstrap_servers: &str) -> String { + format!( + r#" + set -uo pipefail + echo "Starting KRaft voter admission loop against bootstrap servers: {bootstrap_servers}" + while true; do + state=$(curl -s localhost:{metrics_port}/metrics | grep -oE 'kafka_server_raft_metrics_current_state\{{state="[a-z]+"\}}' | grep -oE '"[a-z]+"' | tr -d '"') + if [ "$state" = "observer" ]; then + echo "Local Raft state is observer, attempting add-controller..." + {binary} --bootstrap-controller '{bootstrap_servers}' --command-config {config} add-controller \ + || echo "add-controller attempt failed (this is expected if it already succeeded or a leader election is in progress), will retry" + fi + sleep 10 + done + "#, + bootstrap_servers = bootstrap_servers, + metrics_port = METRICS_PORT, + binary = KAFKA_METADATA_QUORUM_BINARY, + config = ADMIN_CLIENT_PROPERTIES_PATH, + ) +} + +/// The sidecar's `preStop` command: before this controller pod terminates, check that +/// removing it still leaves the quorum with a majority of its *current* voter count, and +/// if so, remove it from the voter set. Always exits 0 — a stuck or failed check must +/// never block pod termination. +/// +/// `node_id` is this controller's own KRaft node id (the same value written to +/// `node.id` in `controller.properties`, derived from `$POD_NAME` and `NODE_ID_OFFSET` +/// exactly as the `kafka` container's own entrypoint does — see `controller_kafka_container_command`). +pub fn quorum_manager_pre_stop_command(bootstrap_servers: &str) -> String { + format!( + r#" + set -uo pipefail + POD_INDEX=$(echo "$POD_NAME" | grep -oE '[0-9]+$') + REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET)) + DEADLINE=$((SECONDS + 25)) + while [ "$SECONDS" -lt "$DEADLINE" ]; do + describe=$({binary} --bootstrap-controller '{bootstrap_servers}' --command-config {config} describe --replication 2>/dev/null) + if [ -n "$describe" ]; then + # NOTE: this parsing was written against the documented `describe --replication` + # tabular output (one voter per line, NodeId as the first column) and must be + # confirmed/adjusted against a live cluster's real output before this is + # considered done -- see Task 4 Step 4 below. + total_voters=$(echo "$describe" | tail -n +2 | grep -c .) + majority=$(( total_voters / 2 + 1 )) + remaining_after_removal=$(( total_voters - 1 )) + if [ "$remaining_after_removal" -ge "$majority" ]; then + directory_id=$(echo "$describe" | tail -n +2 | awk -v id="$REPLICA_ID" '$1 == id {{ print $2 }}') + if [ -n "$directory_id" ]; then + echo "Removing self (node $REPLICA_ID, directory $directory_id) from the voter set..." + {binary} --bootstrap-controller '{bootstrap_servers}' --command-config {config} remove-controller \ + --controller-id "$REPLICA_ID" --controller-directory-id "$directory_id" \ + || echo "remove-controller failed, proceeding with termination anyway" + fi + else + echo "Removing self would break quorum majority ($remaining_after_removal remaining of $majority needed), skipping and retrying..." + fi + break + fi + sleep 2 + done + exit 0 + "#, + bootstrap_servers = bootstrap_servers, + binary = KAFKA_METADATA_QUORUM_BINARY, + config = ADMIN_CLIENT_PROPERTIES_PATH, + ) +} +``` + +Import `METRICS_PORT` at the top of `command.rs` if not already imported (`grep -n "METRICS_PORT" rust/operator-binary/src/controller/build/command.rs`). + +- [ ] **Step 2: Write the failing unit tests for the two command strings** + +Add to (or create) a `#[cfg(test)] mod tests` in `command.rs`: + +```rust +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + assert!(command.contains("--bootstrap-controller 'controller-0:9093,controller-1:9093'")); + assert!(command.contains("add-controller")); + assert!(!command.contains("--bootstrap-controller 'localhost")); + } + + #[test] + fn quorum_manager_pre_stop_command_always_exits_zero() { + let command = quorum_manager_pre_stop_command("controller-0:9093,controller-1:9093"); + assert!(command.trim_end().ends_with("exit 0")); + assert!(command.contains("remove-controller")); + } +} +``` + +If `command.rs` already has a test module, add these two functions inside it instead. + +- [ ] **Step 3: Run the tests to verify they pass** + +Run: `cargo test -p stackable-kafka-operator-binary quorum_manager 2>&1 | tail -40` +Expected: both PASS (these are just string-content assertions, so they should pass immediately once Step 1's functions compile — this is a case where writing the test after the implementation is acceptable, since the "test" here is really a guard against a future accidental typo in the command string, not driving the design). + +- [ ] **Step 4: Build the sidecar container in `statefulset.rs`** + +In `rust/operator-binary/src/controller/build/resource/statefulset.rs`, add a new function near `add_vector_container` (bottom of file): + +```rust +/// Builds the `quorum-manager` sidecar for a controller pod. Returns `None` when this +/// Kafka version doesn't support KIP-853 dynamic quorum tooling, or when Kerberos is +/// enabled (the sidecar's admin-client properties file only covers the TLS/SSL case). +fn build_quorum_manager_container( + resolved_product_image: &ResolvedProductImage, + kafka_security: &ValidatedKafkaSecurity, + quorum_bootstrap_servers: &str, +) -> Result, Error> { + if !supports_dynamic_quorum(&resolved_product_image.product_version) + || kafka_security.has_kerberos_enabled() + { + return Ok(None); + } + + let container_name = "quorum-manager".to_string(); + let mut cb = ContainerBuilder::new(&container_name).context(InvalidContainerNameSnafu { + name: container_name.clone(), + })?; + + cb.image_from_product_image(resolved_product_image) + .command(vec![ + "/bin/bash".to_string(), + "-c".to_string(), + quorum_manager_container_command(quorum_bootstrap_servers), + ]) + .add_env_vars(vec![EnvVar { + name: "POD_NAME".to_string(), + value_from: Some(EnvVarSource { + field_ref: Some(ObjectFieldSelector { + api_version: Some("v1".to_string()), + field_path: "metadata.name".to_string(), + }), + ..EnvVarSource::default() + }), + ..EnvVar::default() + }]) + .resources( + ResourceRequirementsBuilder::new() + .with_cpu_request("100m") + .with_cpu_limit("200m") + .with_memory_request("128Mi") + .with_memory_limit("128Mi") + .build(), + ) + .add_volume_mount(STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) + .context(AddVolumeMountSnafu)? + .lifecycle_pre_stop(LifecycleHandler { + exec: Some(ExecAction { + command: Some(vec![ + "/bin/bash".to_string(), + "-c".to_string(), + quorum_manager_pre_stop_command(quorum_bootstrap_servers), + ]), + }), + ..LifecycleHandler::default() + }); + + Ok(Some(cb.build())) +} +``` + +Add the required imports at the top of `statefulset.rs`: `LifecycleHandler` from `stackable_operator::k8s_openapi::api::core::v1` (alongside the existing `ExecAction`, `EnvVar`, `EnvVarSource`, `ObjectFieldSelector` imports at lines 22-25), `ResourceRequirementsBuilder` (already imported at line 10 for the broker's kcat-prober container — reuse it), and `supports_dynamic_quorum`, `quorum_manager_container_command`, `quorum_manager_pre_stop_command` from `crate::controller::build::{properties, command}`. + +Also add the `Q` sidecar needs the `NODE_ID_OFFSET` env var referenced by its `preStop` script (`$NODE_ID_OFFSET`) — read `statefulset.rs:706-709` (the `kafka` container's own `NODE_ID_OFFSET` env var construction) and add the identical `EnvVar` to the sidecar's `add_env_vars` call in the snippet above, rather than duplicating the whole block — extract the shared computation into a local variable used by both containers if it isn't already. + +Then, inside `build_controller_rolegroup_statefulset`, right after the existing `pod_builder.add_container(kafka_container)` call (around line 579), add: + +```rust + if let Some(quorum_manager_container) = build_quorum_manager_container( + resolved_product_image, + kafka_security, + &quorum_bootstrap_servers, + )? { + pod_builder.add_container(quorum_manager_container); + } +``` + +using the `quorum_bootstrap_servers` binding from Task 3. + +- [ ] **Step 5: Verify the CLI's actual `describe --replication` output shape** + +This step is a real verification action, not a placeholder — the `preStop` script's `awk`/`grep` parsing in Step 1 was written against Kafka's documented tabular format and has not been checked against a live cluster. + +Run: `kubectl exec -n test-kafka-controller-default-0 -c kafka -- /stackable/kafka/bin/kafka-metadata-quorum.sh --bootstrap-controller :9093 --command-config /stackable/config/admin-client.properties describe --replication` + +(This requires Task 2's `admin-client.properties` to already be deployed — run this verification after Tasks 2-4 are all merged into a real running cluster, e.g. via a manual `./scripts/run-tests` smoke-kraft run, before considering this task done.) Compare the real column layout (which column holds `NodeId`, which holds `DirectoryId`) against the `awk -v id="$REPLICA_ID" '$1 == id { print $2 }'` assumption in Step 1, and adjust the column indices in `quorum_manager_pre_stop_command` if they don't match. Re-run the unit tests from Step 3 after any change (they assert command *structure*, not the exact awk column numbers, so they should still pass, but re-run them anyway to be safe). + +- [ ] **Step 6: Build and run the full test suite** + +Run: `cargo build -p stackable-kafka-operator-binary 2>&1 | tail -60` +Run: `cargo test -p stackable-kafka-operator-binary 2>&1 | tail -80` +Expected: builds cleanly, all tests PASS (this also exercises every existing `statefulset.rs`/`config_map.rs` test, confirming the new sidecar doesn't break broker-pod builds, which must never get this container). + +- [ ] **Step 7: Regenerate CRDs and check for unexpected diffs** + +Run: `make regenerate-charts 2>&1 | tail -40` +Expected: no diff, since this task adds a container by string literal name rather than a new `ContainerName`-enum variant, so the CRD's `logging.containers` schema is unchanged. If `make regenerate-charts` produces an unexpected diff, investigate before proceeding — it likely means a CRD-visible type changed somewhere in this task's diff. + +- [ ] **Step 8: Commit** + +```bash +git add rust/operator-binary/src/controller/build/resource/statefulset.rs rust/operator-binary/src/controller/build/command.rs +git commit -m "feat: add quorum-manager sidecar to controller pods + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 5: Switch controller StatefulSet to `OrderedReady` pod management + +**Files:** + +- Modify: `rust/operator-binary/src/controller/build/resource/statefulset.rs` + +**Interfaces:** + +- Consumes: nothing new. +- Produces: `POD_MANAGEMENT_POLICY_ORDERED_READY` constant, consumed only by this task's own change and asserted by Task 6's unit test. + +- [ ] **Step 1: Write the failing unit test** + +In the `statefulset.rs` test module (created in Task 3 or already present), add: + +```rust + #[test] + fn controller_statefulset_uses_ordered_ready_pod_management() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller StatefulSet is built"); + + assert_eq!( + sts.spec.expect("the StatefulSet has a spec").pod_management_policy, + Some("OrderedReady".to_string()) + ); + } + + #[test] + fn broker_statefulset_still_uses_parallel_pod_management() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-broker-default")) + .expect("the broker StatefulSet is built"); + + assert_eq!( + sts.spec.expect("the StatefulSet has a spec").pod_management_policy, + Some("Parallel".to_string()) + ); + } +``` + +This uses a `kraft_mode_cluster()` fixture. Task 3 deliberately deferred creating this fixture (see its Step 1) in favor of this task owning it. **This task creates `kraft_mode_cluster()`** in this file's test module, copied from the pattern shown in the exploration: a minimal `KafkaCluster` YAML with `clusterConfig.metadataManager: kraft`, one controller role group of 3 replicas, one broker role group of 3 replicas, resolved via `crate::controller::test_support::{minimal_kafka, validated_cluster}`. + +- [ ] **Step 2: Run the tests to verify the controller one fails** + +Run: `cargo test -p stackable-kafka-operator-binary pod_management 2>&1 | tail -30` +Expected: `broker_statefulset_still_uses_parallel_pod_management` PASSes (no change yet), `controller_statefulset_uses_ordered_ready_pod_management` FAILs (`Parallel` != `OrderedReady`). + +- [ ] **Step 3: Make the change** + +In `build_controller_rolegroup_statefulset`, add a new constant near the existing `POD_MANAGEMENT_POLICY_PARALLEL` (`statefulset.rs:127`): + +```rust +const POD_MANAGEMENT_POLICY_ORDERED_READY: &str = "OrderedReady"; +``` + +And change the controller `StatefulSetSpec` construction (`statefulset.rs:620`) from: + +```rust + pod_management_policy: Some(POD_MANAGEMENT_POLICY_PARALLEL.to_string()), +``` + +to: + +```rust + pod_management_policy: Some(POD_MANAGEMENT_POLICY_ORDERED_READY.to_string()), +``` + +Leave the broker StatefulSet's construction (`statefulset.rs:435`) unchanged — it must keep using `POD_MANAGEMENT_POLICY_PARALLEL`. + +- [ ] **Step 4: Run the tests to verify they pass** + +Run: `cargo test -p stackable-kafka-operator-binary pod_management 2>&1 | tail -30` +Expected: both PASS. + +- [ ] **Step 5: Commit** + +```bash +git add rust/operator-binary/src/controller/build/resource/statefulset.rs +git commit -m "feat: use OrderedReady pod management for controller StatefulSets + +Serializes scale-down so each controller's preStop hook (self-removal +from the KRaft voter set) completes before the next pod terminates. + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 6: Unit tests for sidecar presence/absence and version gating + +**Files:** + +- Modify: `rust/operator-binary/src/controller/build/resource/statefulset.rs` + +**Interfaces:** + +- Consumes: `kraft_mode_cluster()` fixture (created by Task 5 — Task 3 deliberately deferred it), `build_quorum_manager_container` / the sidecar's presence in the built `StatefulSet` (Task 4), the `kerberos()` security fixture from `security.rs`'s test module (Task 2 — may need its visibility bumped to `pub(crate)` for this task to reach it). + +- [ ] **Step 1: Write the failing tests** + +```rust + fn controller_containers( + cluster: &crate::controller::ValidatedCluster, + ) -> Vec { + let resources = crate::controller::build::build(cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller StatefulSet is built"); + sts.spec + .expect("the StatefulSet has a spec") + .template + .spec + .expect("the pod template has a spec") + .containers + } + + #[test] + fn controller_pods_get_a_quorum_manager_sidecar_on_supported_versions() { + let cluster = kraft_mode_cluster(); + let containers = controller_containers(&cluster); + + assert!( + containers.iter().any(|c| c.name == "quorum-manager"), + "expected a quorum-manager sidecar, got containers: {:?}", + containers.iter().map(|c| &c.name).collect::>() + ); + } + + #[test] + fn quorum_manager_sidecar_targets_bootstrap_servers_in_its_command() { + let cluster = kraft_mode_cluster(); + let containers = controller_containers(&cluster); + let sidecar = containers + .iter() + .find(|c| c.name == "quorum-manager") + .expect("the quorum-manager sidecar is built"); + + let command = sidecar + .command + .as_ref() + .expect("the sidecar has a command") + .join(" "); + assert!(command.contains("add-controller")); + + let pre_stop_command = sidecar + .lifecycle + .as_ref() + .and_then(|l| l.pre_stop.as_ref()) + .and_then(|h| h.exec.as_ref()) + .and_then(|e| e.command.as_ref()) + .expect("the sidecar has a preStop exec hook") + .join(" "); + assert!(pre_stop_command.contains("remove-controller")); + assert!(pre_stop_command.trim_end().ends_with("exit 0")); + } + + #[test] + fn controller_pods_get_no_quorum_manager_sidecar_on_kafka_3_7() { + let kafka = crate::controller::test_support::minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.7.2 + clusterConfig: + metadataManager: kraft + controllers: + roleGroups: + default: + replicas: 3 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + let cluster = crate::controller::test_support::validated_cluster(&kafka); + let containers = controller_containers(&cluster); + + assert!(!containers.iter().any(|c| c.name == "quorum-manager")); + } + + #[test] + fn controller_pods_get_no_quorum_manager_sidecar_when_kerberos_is_enabled() { + // This is a Global Constraint (see the plan header): the sidecar's admin-client + // properties file only covers the TLS/SSL case, so it must never be added when + // Kerberos is enabled, even on an otherwise-supported Kafka version. + // + // Rather than building a full CRD-level Kerberos fixture (which needs a resolved + // AuthenticationClass threaded through `DereferencedObjects`, more than this test + // needs), call `build_quorum_manager_container` directly — it already takes + // `&ValidatedKafkaSecurity` as a parameter, so a fixture at that level is enough. + // Reuse the `kerberos()` fixture from `security.rs`'s existing test module (see + // Task 2) for a security value with Kerberos enabled; import it, adjusting its + // visibility to `pub(crate)` in `security.rs` if it is not already visible here. + let cluster = kraft_mode_cluster(); + let kerberos_security = crate::controller::build::security::tests::kerberos(); + + let result = build_quorum_manager_container( + &cluster.image, + &kerberos_security, + "controller-0:9093", + ) + .expect("build_quorum_manager_container does not error for a kerberos security value"); + + assert!(result.is_none()); + } + + #[test] + fn broker_pods_never_get_a_quorum_manager_sidecar() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-broker-default")) + .expect("the broker StatefulSet is built"); + let containers = sts + .spec + .expect("the StatefulSet has a spec") + .template + .spec + .expect("the pod template has a spec") + .containers; + + assert!(!containers.iter().any(|c| c.name == "quorum-manager")); + } +``` + +Verify `productVersion: 3.7.2` is actually a version accepted by this repo's product-version validation (some operators restrict to an exact known list) — check: + +Run: `grep -rn "3.7" rust/crd/src/ tests/test-definition.yaml 2>/dev/null | head -20` + +If `3.7.2` isn't a recognized version, use whatever 3.7.x version is used elsewhere in this repo's own tests/fixtures instead. + +- [ ] **Step 2: Run the tests to verify they fail (or pass, if Task 4/5 already got this right)** + +Run: `cargo test -p stackable-kafka-operator-binary quorum_manager 2>&1 | tail -60` + +If Tasks 4-5 were implemented correctly, these should already PASS since they're testing behavior those tasks already built — this task exists to lock that behavior in with explicit regression coverage, not to drive new implementation. If any fail, fix the implementation in `statefulset.rs` from Task 4/5 (not the test) unless the test itself has a mistaken assumption — re-read Task 4/5's code before changing either. + +- [ ] **Step 3: Run the full test suite one more time** + +Run: `cargo test -p stackable-kafka-operator-binary 2>&1 | tail -80` +Expected: all PASS. + +- [ ] **Step 4: Commit** + +```bash +git add rust/operator-binary/src/controller/build/resource/statefulset.rs +git commit -m "test: cover quorum-manager sidecar presence, version gate, and commands + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 7: Fix and strengthen the kuttl scale-up/scale-down assertions + +**Files:** + +- Modify: `tests/templates/kuttl/operations-kraft/60-assert.yaml.j2` +- Modify: `tests/templates/kuttl/operations-kraft/70-assert.yaml.j2` + +**Interfaces:** none (test-only, no Rust interfaces). + +Both files currently have a real bug (found during design exploration): the two YAML documents for the broker and controller `StatefulSet` assertions are missing a `---` separator between them, which likely means the second document (the controller assertion) is silently ignored by the YAML parser or produces unexpected behavior. This task fixes that bug and adds a voter-count check. + +- [ ] **Step 1: Read both files in full** + +Run: `cat tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 tests/templates/kuttl/operations-kraft/70-assert.yaml.j2` + +Confirm the missing `---` before the second `apiVersion: apps/v1` block in each file. + +- [ ] **Step 2: Fix the missing document separator in `60-assert.yaml.j2`** + +Insert a `---` line immediately before the second `apiVersion: apps/v1` (the `test-kafka-controller-default` StatefulSet assertion), so the file has three `---`-separated documents: the `TestAssert` header/commands block, the broker StatefulSet assertion, and the controller StatefulSet assertion. + +- [ ] **Step 3: Apply the identical fix to `70-assert.yaml.j2`** + +Same change, same reasoning. + +- [ ] **Step 4: Add a voter-count assertion command to `60-assert.yaml.j2`** + +In the `commands:` list of the `TestAssert` document (alongside the existing `kubectl -n $NAMESPACE wait --for=condition=available ...` command), add: + +```yaml + - script: | + kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ + /stackable/kafka/bin/kafka-metadata-quorum.sh \ + --bootstrap-controller test-kafka-controller-default-0.test-kafka-controller-default-headless.$NAMESPACE.svc.cluster.local:9093 \ + --command-config /stackable/config/admin-client.properties \ + describe --replication | tail -n +2 | wc -l | grep -q '^5$' +``` + +Verify the exact headless service name pattern (`test-kafka-controller-default-headless`) against how the FQDN is actually constructed elsewhere in this test suite — grep other files in `tests/templates/kuttl/operations-kraft/` for an existing `--bootstrap-server`/FQDN reference to copy the exact naming convention rather than guessing it: + +Run: `grep -rn "headless\|bootstrap-server" tests/templates/kuttl/operations-kraft/*.j2 tests/templates/kuttl/smoke-kraft/*.j2 2>/dev/null | head -20` + +Adjust the hostname in the command above to match whatever convention those files actually use. + +- [ ] **Step 5: Add the equivalent assertion to `70-assert.yaml.j2`, expecting 3 voters** + +Same command, with `grep -q '^3$'` instead of `'^5$'`, matching the scaled-down replica count. + +- [ ] **Step 6: Run the kuttl test manually** + +Run: `./scripts/run-tests --skip-release --test operations-kraft_kafka-kraft-4.2.1_openshift-false 2>&1 | tail -100` + +Expected: PASS, including the new voter-count checks in steps 60 and 70. If the voter-count check fails while the StatefulSet readiness check passes, that's a real signal the sidecar (Task 4) isn't actually admitting/removing voters correctly — go back to Task 4 and debug using the same `vector tap` / `kubectl logs -c quorum-manager` techniques, rather than loosening this assertion. + +- [ ] **Step 7: Commit** + +```bash +git add tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 +git commit -m "test: fix missing YAML separator and assert voter count in scale tests + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 8: Update documentation and remove the "unsupported" claim + +**Files:** + +- Modify: `docs/modules/kafka/pages/usage-guide/kraft-controller.adoc` +- Modify: `tests/templates/kuttl/operations-kraft/README.md` +- Modify: `CHANGELOG.md` + +**Interfaces:** none. + +- [ ] **Step 1: Update `kraft-controller.adoc`** + +Read the file in full (already read during brainstorming). Remove or rewrite the "Scaling controller replicas up is not supported" bullet under "Known Issues" and the entire "Scaling issues" subsection under "Troubleshooting", replacing them with a short description of the new behavior: + +- Controllers can now be scaled up and down on a running cluster. +- A per-pod `quorum-manager` sidecar handles admitting/removing the pod from the KRaft voter set. +- This requires a Kafka version that supports KIP-853 dynamic quorum tooling (everything except `3.7.x`). +- Scaling more than one controller down at a time is processed one pod at a time (`OrderedReady`), not in parallel. +- If a `remove-controller` call fails during pod termination (e.g. no reachable leader within the grace period), a stale voter entry can be left behind and requires manual cleanup — state this as a known limitation, do not imply it's fully automatic in every case. + +Also update the "Internal operator details" bullet that currently says "the operator does not perform the follow-up step required to add a controller to an already-formed quorum's voter set" — this is no longer true. + +- [ ] **Step 2: Update the kuttl README** + +`tests/templates/kuttl/operations-kraft/README.md` currently states "Scaling controllers from 3 -> 1 doesn't work. Both brokers and controllers try to communicate with old controllers." Verify whether this specific limitation (scaling below a certain floor) is still expected to hold after this change — if scaling controllers down to 1 was never exercised by these tests (they only go 3→5→3), leave this caveat in place rather than removing an unverified claim; do not claim a scenario is fixed that this plan's tests don't actually cover. + +- [ ] **Step 3: Add the CHANGELOG entry** + +In `CHANGELOG.md`, insert a new `### Added` section between `## [Unreleased]` and the existing `### Changed` section: + +```markdown +### Added + +- KRaft controller replicas can now be scaled up and down on a running cluster: a new + `quorum-manager` sidecar container on each controller pod admits itself into the KRaft + voter set on startup and removes itself before termination ([#NNNN]). +``` + +Add the corresponding link reference at the bottom of the file, in ascending numeric order alongside the existing `[#985]`/`[#990]`/etc. links: + +```markdown +[#NNNN]: https://github.com/stackabletech/kafka-operator/pull/NNNN +``` + +Leave `NNNN` as a literal placeholder for the real PR number — fill it in when the PR is actually opened (this is the one acceptable use of a placeholder in this plan, since the number doesn't exist until the PR is created; every other file in this plan has zero placeholders). + +- [ ] **Step 4: Commit** + +```bash +git add docs/modules/kafka/pages/usage-guide/kraft-controller.adoc tests/templates/kuttl/operations-kraft/README.md CHANGELOG.md +git commit -m "docs: document KRaft controller scale-up/down support + +Co-Authored-By: Claude Sonnet 5 " +``` + +--- + +## Task 9: Full verification gate + +**Files:** none (verification only). + +- [ ] **Step 1: Full build** + +Run: `cargo build --workspace 2>&1 | tail -60` +Expected: clean build. + +- [ ] **Step 2: Full test suite** + +Run: `cargo test --workspace 2>&1 | tail -100` +Expected: all PASS. + +- [ ] **Step 3: Clippy** + +Run: `cargo clippy --all-targets -- -D warnings 2>&1 | tail -100` +Expected: no warnings/errors. Fix anything that comes up before proceeding. + +- [ ] **Step 4: Format check** + +Run: `cargo fmt --check 2>&1 | tail -60` +Expected: no diff. If there is one, run `cargo fmt` and amend the relevant task's commit. + +- [ ] **Step 5: Regenerate charts/CRDs one final time** + +Run: `make regenerate-charts 2>&1 | tail -60` +Expected: no diff (confirmed already in Task 4, re-checked here after all subsequent tasks in case anything else drifted). + +- [ ] **Step 6: Full kuttl run for the affected test suite** + +Run: `./scripts/run-tests --skip-release --test operations-kraft_kafka-kraft-4.2.1_openshift-false 2>&1 | tail -150` +Run: `./scripts/run-tests --skip-release --test operations-kraft_kafka-kraft-3.9.2_openshift-false 2>&1 | tail -150` +Expected: both PASS. + +- [ ] **Step 7: Commit any fixups from this task as a single commit, if any were needed** + +```bash +git add -A +git commit -m "chore: fix clippy/fmt findings from verification pass + +Co-Authored-By: Claude Sonnet 5 " +``` + +If nothing needed fixing, skip this commit — don't create an empty one. diff --git a/docs/superpowers/specs/2026-08-14-kraft-dynamic-voter-membership-design.md b/docs/superpowers/specs/2026-08-14-kraft-dynamic-voter-membership-design.md new file mode 100644 index 00000000..d7c52714 --- /dev/null +++ b/docs/superpowers/specs/2026-08-14-kraft-dynamic-voter-membership-design.md @@ -0,0 +1,197 @@ +# KRaft dynamic voter membership (scale-up / scale-down) + +Status: approved for planning +Date: 2026-08-14 +Branch this was designed on: `main` @ `5211842` + +## Problem + +Apache Kafka's KRaft dynamic quorum (KIP-853) requires an explicit +follow-up step to change the voter set of an already-formed quorum: +`kafka-metadata-quorum.sh add-controller` to admit a new controller, +`remove-controller` to retire one. The Stackable Kafka operator +currently only performs the one-time `--initial-controllers` step at +`kafka-storage.sh format` time. Any controller pod added after initial +cluster formation registers itself and starts up, but never leaves the +Raft `observer` state — it can never become `leader`/`follower`/`voted`, +so it never becomes healthy, and there is no supported way to remove a +controller from the voter set either. This is documented today as a +flat "do not scale controller replicas on a running cluster" limitation +in `docs/modules/kafka/pages/usage-guide/kraft-controller.adoc`. + +Goal: let `spec.controllers.roleGroups..replicas` be scaled up +and down on a running cluster, with the voter set kept in sync +automatically. + +## Scope + +- In scope: admitting new controllers to the voter set on scale-up, + removing controllers from the voter set on scale-down, for live + replica-count changes on an already-formed cluster. +- Out of scope: whole-cluster graceful-deletion draining (a + finalizer-based mechanism is a separate concern from live replica + changes and is not addressed here). ZooKeeper-to-KRaft migration is + unaffected. Kerberos support for KRaft is unaffected. +- This design was developed independently of, and does not reuse code + from, any prior unmerged work on a KRaft "quorum health gate" or + graceful-teardown finalizer — that work was discarded (local + branches deleted) before this design was started. + +## Non-goals / explicitly rejected approaches + +- **Operator-side kube-exec.** An earlier direction had the operator + itself exec into pods (`pods/exec`) to run + `kafka-metadata-quorum.sh`, gated by new reconcile phases (a + pre-`build` gate clamping the effective replica count on scale-down, + a post-`apply` phase admitting new voters on scale-up). This was + rejected in favor of the sidecar approach below: it needed new RBAC, + a new "live cluster" client capability the operator has never had, + and two new reconcile phases, none of which are needed once the pods + manage their own membership. +- **`controller.quorum.auto.join.enable`.** Delegates scale-up + self-promotion entirely to Kafka with zero new operator capability, + but doesn't address scale-down at all (still needs an active + `remove-controller` step), and gives up visibility into *why* + admission might be stuck. Not chosen because scale-down still needs + the same sidecar mechanism anyway, so this would only save the + add-controller half of the problem while adding a version dependency + to check. + +## Design + +### Architecture + +The operator gains **no new awareness of live quorum state**. The +existing reconcile pipeline (`dereference → validate → build → apply → +update_status`) is untouched. All quorum membership management is +delegated to the controller pods themselves, via: + +1. A new sidecar container, controller-role-only, reusing the `kafka` + product image (so `kafka-metadata-quorum.sh` and the TLS trust + material already mounted for the `kafka` container are available + without new volumes). +2. A `preStop` lifecycle hook on that sidecar. +3. `podManagementPolicy: OrderedReady` on the controller StatefulSet + (currently `Parallel`). + +Both the sidecar and the `preStop` hook are only added for Kafka +versions that support KIP-853 dynamic quorum tooling — mirrors the +existing per-version special-casing already present around +`--initial-controllers` for 3.7.x. Older versions get no sidecar at +all and keep today's documented "unsupported" behavior. + +### Components + +**Add-loop script** (the sidecar's main process, runs for the pod's +whole lifetime): + +- Polls the local JMX Prometheus metrics endpoint (the same + `kafka_server_raft_metrics_current_state` series the existing + readiness probe already reads) on a short interval. +- While state is `observer`, runs + `kafka-metadata-quorum.sh add-controller` against + `controller.quorum.bootstrap.servers` (this must be invoked locally + on the joining node — it reads local KRaft directory state + automatically, which also sidesteps the fake placeholder directory-id + used by `KafkaPodDescriptor::as_voter()` at format time; that + placeholder was flagged during design exploration as a hazard for + any tooling that validates directory ids, but `add-controller` does + not consume it). +- Treats "already a voter" responses as success and keeps polling at + the same interval indefinitely (cheap, idempotent, self-healing — + no persisted state, no operator involvement). + +This also resolves what looked like a circular dependency during +design: the existing readiness probe can only pass once raft state +leaves `observer`, so gating admission on pod-readiness would be +circular. The sidecar's loop is independent of the pod's own readiness +state, so there is no cycle. + +**Remove script** (the sidecar's `preStop` hook, runs once at +termination): + +1. Runs `kafka-metadata-quorum.sh describe --replication` to get the + current voter list. +2. Checks that removing itself would still leave a majority of the + *pre-removal* voter count. This check is done explicitly by the + script — the design does not assume `remove-controller` refuses an + unsafe removal on Kafka's side. +3. If safe, calls `remove-controller` for itself. +4. The whole hook is bounded by a timeout comfortably inside + `terminationGracePeriodSeconds`, and always exits `0` — a stuck or + failed check must never block pod termination indefinitely. + +### Data flow + +**Scale-up:** an ordinary declarative replica increase on the +controller StatefulSet (no change from today) creates a new pod. Its +`kafka` container boots exactly as today (format + start). Its sidecar +independently loops until it observes itself admitted. The existing +readiness probe starts passing once raft state leaves `observer`. If +multiple controllers are added at once, each pod's sidecar self-admits +independently; Kafka's leader serializes the actual `AddVoter` +application, so no operator-side coordination is required. + +**Scale-down:** an ordinary declarative replica decrease (no change +from today). `OrderedReady` means Kubernetes terminates exactly the +highest-ordinal pod, runs its `preStop` hook (self-removal via the +script above), and waits for full termination before considering the +next pod — this is what gives one-at-a-time, majority-checked draining +for a decrease of any size, entirely via a StatefulSet setting. No +Rust-side "gate the effective replica count" logic is needed. + +### Error handling + +- Transient `add-controller` / `describe` failures (e.g. a leader + election in flight) are simply retried by the loop on its normal + interval. There is no alerting path today: the sidecar has no + Kubernetes API access by design (that's the point — no new RBAC), + so failures are visible only via `kubectl logs` on the sidecar + container. +- **Known observability gap:** the sidecar's stdout will *not* be + picked up by the existing vector log-aggregation pipeline, which + only tails structured `*.log4j.xml` / `*.log4j2.xml` files written + by the JVM's own logging config (confirmed by direct inspection of + the deployed `vector.yaml` ConfigMaps during an unrelated + investigation). This is a real, known limitation of this design, not + something papered over — a future iteration could have the sidecar + write structured lines to a file under the shared log directory to + get picked up, but that is not included in this design's initial + scope. +- `preStop` removal timing out or failing (e.g. no reachable leader + within the grace period): the pod still terminates on schedule. A + stale voter entry can be left behind in the quorum in that case. + This is a genuine, stated limitation — recovery in that scenario is + manual (the same "no supported automated path" caveat that already + exists in the current docs for quorum-reconfiguration edge cases). + +### Testing + +- The existing `tests/templates/kuttl/operations-kraft/60-scale-controller-up.yaml.j2` + / `70-scale-controller-down.yaml.j2` kuttl tests (active on `main`, + not disabled) should pass once this is implemented. Strengthen their + `*-assert.yaml.j2` counterparts beyond "StatefulSet reports N/N + ready" to also verify voter count matches replica count post-scale + (e.g. via `describe --replication` run from the test's `python-0` + pod), so the test catches a silently-stuck-in-`observer` regression, + not just a stuck-not-ready one. +- Unit tests in `rust/operator-binary/src/controller/build/resource/statefulset.rs`, + mirroring the existing probe tests added in `ec59dab`: + - the sidecar container is present only on the controller role, and + only for Kafka versions that support dynamic quorum tooling; + - the sidecar's `preStop` command matches the expected removal + script invocation; + - the controller StatefulSet's `podManagementPolicy` is + `OrderedReady`. + +## Open questions for implementation planning + +- Exact minimum Kafka version for the version gate (needs verification + against Kafka's own KIP-853 tooling maturity, not assumed here). +- Exact script implementation (shell, embedded via ConfigMap vs. an + inline `bash -c` command similar to the existing probe commands in + `statefulset.rs`) and its `--command-config` security settings + (matching whatever TLS/SASL configuration the `kafka` container + already uses for its internal listener). +- Whether to close the sidecar-log observability gap noted above as + part of this work or as explicit follow-up. diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 4805a623..dcb3f8b6 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -11,7 +11,7 @@ use super::properties::ConfigFileName; use crate::{ controller::{build::security::copy_opa_tls_cert_command, security::ValidatedKafkaSecurity}, crd::{ - BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, STACKABLE_CONFIG_DIR, + BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, METRICS_PORT, STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR, }, }; @@ -37,12 +37,27 @@ pub fn kafka_log_opts_env_var() -> String { "KAFKA_LOG4J_OPTS".to_string() } +/// Shell snippet setting `$POD_INDEX` to this pod's ordinal, parsed from the trailing digits +/// of `$POD_NAME` (e.g. `2` for `..-controller-default-2`). +/// +/// Paired with [`EXPORT_REPLICA_ID`] (see there for why the split): used, in some combination, +/// by four call sites that used to each duplicate this derivation with slightly drifted +/// whitespace — the broker and controller `kafka` containers' own entrypoints, and the +/// `quorum-manager` sidecar's main loop and `preStop` hook. +const DERIVE_POD_INDEX: &str = r#"POD_INDEX=$(echo "$POD_NAME" | grep -oE '[0-9]+$')"#; + +/// Shell snippet exporting `$REPLICA_ID` (this container's KRaft node id) from `$POD_INDEX` +/// (see [`DERIVE_POD_INDEX`], which must run first) and `$NODE_ID_OFFSET`. Exported (rather +/// than a plain assignment) because every caller either runs `config-utils template` or the +/// `quorum-manager` sidecar's `kafka-metadata-quorum.sh`/`curl` calls as a *subprocess*, which +/// need `REPLICA_ID` in their environment, not just this shell's. +const EXPORT_REPLICA_ID: &str = "export REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET))"; + /// Returns the commands to start the main Kafka container pub fn broker_kafka_container_commands( kraft_mode: bool, controller_descriptors: Vec, kafka_security: &ValidatedKafkaSecurity, - product_version: &str, ) -> String { formatdoc! {" {COMMON_BASH_TRAP_FUNCTIONS} @@ -65,18 +80,17 @@ pub fn broker_kafka_container_commands( false => "".to_string(), }, import_opa_tls_cert = copy_opa_tls_cert_command(kafka_security), - broker_start_command = broker_start_command(kraft_mode, controller_descriptors, product_version), + broker_start_command = broker_start_command(kraft_mode, controller_descriptors), } } fn broker_start_command( kraft_mode: bool, controller_descriptors: Vec, - product_version: &str, ) -> String { let common_command = formatdoc! {" - export POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$') - export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET)) + {derive_pod_index} + {export_replica_id} if [ -f \"{broker_id_pod_map_dir}/$POD_NAME\" ]; then REPLICA_ID=$(cat \"{broker_id_pod_map_dir}/$POD_NAME\") @@ -88,6 +102,8 @@ fn broker_start_command( cp {config_dir}/{jaas_file} /tmp/{jaas_file} config-utils template /tmp/{jaas_file} ", + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, broker_id_pod_map_dir = BROKER_ID_POD_MAP_DIR, config_dir = STACKABLE_CONFIG_DIR, properties_file = ConfigFileName::BrokerProperties, @@ -102,7 +118,7 @@ fn broker_start_command( bin/kafka-server-start.sh /tmp/{properties_file} & ", properties_file = ConfigFileName::BrokerProperties, - initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), + initial_controller_command = initial_controllers_command(&controller_descriptors), } } else { formatdoc! {" @@ -158,7 +174,6 @@ wait_for_termination() pub fn controller_kafka_container_command( controller_descriptors: Vec, - product_version: &str, ) -> String { formatdoc! {" {BASH_TRAP_FUNCTIONS} @@ -166,8 +181,8 @@ pub fn controller_kafka_container_command( prepare_signal_handlers containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop & - POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$') - export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET)) + {derive_pod_index} + {export_replica_id} cp {config_dir}/{properties_file} /tmp/{properties_file} @@ -180,13 +195,213 @@ pub fn controller_kafka_container_command( {create_vector_shutdown_file_command} ", remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR), + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, config_dir = STACKABLE_CONFIG_DIR, properties_file = ConfigFileName::ControllerProperties, - initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), + initial_controller_command = initial_controllers_command(&controller_descriptors), create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) } } +/// The `kafka-metadata-quorum.sh` binary, referenced by its absolute path (matching every +/// other exec-into-pod usage of a Kafka CLI tool in this repo, e.g. the kuttl test scripts +/// under `tests/templates/kuttl/*/*.sh`), rather than the relative `bin/...` form used only +/// inside the `kafka` container's own entrypoint (which runs with the Kafka install dir as +/// its working directory). +const KAFKA_METADATA_QUORUM_BINARY: &str = "/stackable/kafka/bin/kafka-metadata-quorum.sh"; + +const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/stackable/config/admin-client.properties"; + +/// The merged config used only for `add-controller` (self-registration). +/// +/// `add-controller` is not a plain admin-client call: the same process that connects to the +/// quorum also reads `node.id` and its own `listeners`/`controller.listener.names` from the +/// **same** `--command-config` file to build the voter registration payload (confirmed live: +/// pointed at the plain [`ADMIN_CLIENT_PROPERTIES_PATH`], every attempt failed with `node.id +/// not found in configuration file`, so no controller was ever able to admit itself as a +/// voter). But that rendered `controller.properties` has no bare `security.protocol`/`ssl.*` +/// keys of its own — only the `listener.name..ssl.*`-prefixed ones the server process +/// uses for its listeners — so using it *instead of* the admin-client config leaves the +/// AdminClient with no TLS config and unable to reach the (TLS-only) bootstrap controller. +/// Concatenating both files (also confirmed live) gives `add-controller` everything it reads: +/// the bare `ssl.*`/`security.protocol` keys for its own connection, plus `node.id` and the +/// listener keys for the registration payload. +/// +/// **Order matters.** There is no key overlap between the two files today, but +/// `controller.properties` accepts unconditional `configOverrides` merged into it (see +/// `controller_properties::build`), so a user override there could add a colliding key. Java +/// properties parsing lets a later occurrence of the same key win, so `controller.properties` +/// is concatenated *first* and [`ADMIN_CLIENT_PROPERTIES_PATH`] *last* — that way the client +/// TLS config `add-controller` connects with always wins by construction, rather than +/// depending on there being no collision today. +const ADD_CONTROLLER_PROPERTIES_PATH: &str = "/tmp/add-controller.properties"; + +/// Wall-clock bound (seconds) applied to every individual `kafka-metadata-quorum.sh` +/// invocation via `timeout`. The Java AdminClient can otherwise retry internally for far +/// longer than any of this file's own script-level deadlines, which matters most in +/// `quorum_manager_pre_stop_command`: it runs exactly when peers may be unreachable, and a +/// hung admin-client call there would burn into `terminationGracePeriodSeconds` (default: +/// 30 minutes) rather than the script's own 25s budget. +const CLI_CALL_TIMEOUT_SECONDS: u32 = 15; + +/// The sidecar's main-loop command: while this controller's local Raft state is +/// `observer`, repeatedly attempt to admit it into the quorum's voter set. +/// +/// `bootstrap_servers` is the comma-joined `host:port` list produced by +/// `kraft_controllers(...)` (see `build/properties/mod.rs`). +/// +/// Explicitly traps `TERM` and exits: this script runs as the container's PID 1, and the +/// kernel suppresses the default action of unhandled signals for PID 1, so without this +/// trap the loop below would never notice `SIGTERM` and would run until Kubernetes gives up +/// waiting and sends `SIGKILL` after the full `terminationGracePeriodSeconds` (confirmed +/// live: with no trap, this container kept looping — and its pod kept reporting as +/// `Terminating` — long after the `kafka` container in the same pod had shut down +/// gracefully). The `sleep 10 &`/`wait $!` pair (rather than a plain `sleep 10`) lets the +/// trap fire immediately: bash's `wait` builtin is interrupted as soon as a trapped signal +/// arrives, whereas a foreground `sleep` would only be noticed once it finished. +/// +/// Renders [`ADD_CONTROLLER_PROPERTIES_PATH`] once at startup (this controller's identity +/// and listener address don't change for the container's lifetime) by reusing the same +/// `$POD_NAME`/`NODE_ID_OFFSET` → `REPLICA_ID` derivation ([`DERIVE_POD_INDEX`]/ +/// [`EXPORT_REPLICA_ID`]), and the same `config-utils template` render step, as the `kafka` +/// container's own entrypoint (see [`controller_kafka_container_command`]) — see +/// [`ADD_CONTROLLER_PROPERTIES_PATH`] for why `add-controller` needs this merged file rather +/// than the plain admin-client config, and for why the concatenation order matters. +/// +/// The render/merge preamble's inputs are static, operator-rendered config (env vars set +/// once at pod creation), so a failure there is a genuine misconfiguration that retrying +/// won't fix. It must still be loud in the logs, but it must *not* crash the container: a +/// container with no readiness probe is only `Ready` while `Running`, and (with +/// `OrderedReady` pod management on every non-Kerberos controller `StatefulSet`) a +/// crash-looping sidecar would make its whole pod `NotReady` and block scale/update +/// progress for every sibling pod in the role, not just the broken one. So on failure this +/// falls into a "degraded" loop that repeats a clear error every 30s and never attempts +/// `add-controller` (there is no valid rendered config to use), keeping the container alive +/// and `Running` while the problem stays visible via `kubectl logs`. This deliberately does +/// *not* retry the render/merge step itself — that would look like it might eventually +/// succeed, when the actual cause is a misconfiguration that only a human or a new rollout +/// can fix. +pub fn quorum_manager_container_command(bootstrap_servers: &str) -> String { + format!( + r#" + set -uo pipefail + trap 'exit 0' TERM + {derive_pod_index} + [ -n "$POD_INDEX" ] || exit 0 + {export_replica_id} + + if cp {config_dir}/{controller_properties_file} /tmp/{controller_properties_file} \ + && config-utils template /tmp/{controller_properties_file} \ + && cat /tmp/{controller_properties_file} {admin_client_config} > {add_controller_config}; then + echo "Starting KRaft voter admission loop against bootstrap servers: {bootstrap_servers}" + while true; do + state=$(curl -s --max-time 5 --connect-timeout 2 localhost:{metrics_port}/metrics | grep -oE 'kafka_server_raft_metrics_current_state\{{state="[a-z]+"\}}' | grep -oE '"[a-z]+"' | tr -d '"') + if [ "$state" = "observer" ]; then + echo "Local Raft state is observer, attempting add-controller..." + timeout {cli_timeout} {binary} --bootstrap-controller '{bootstrap_servers}' --command-config {add_controller_config} add-controller \ + || echo "add-controller attempt failed (this is expected if it already succeeded or a leader election is in progress), will retry" + elif [ -z "$state" ]; then + echo "Could not determine local Raft state (metrics scrape returned nothing), will retry" + else + echo "Local Raft state is '$state', nothing to do" + fi + sleep 10 & + wait $! + done + else + echo "ERROR: quorum-manager failed to render or merge its configuration (see errors above); this looks like a genuine misconfiguration, not a transient failure." + while true; do + echo "ERROR: quorum-manager is degraded and will NOT attempt add-controller: configuration render/merge failed at startup and this container is not retrying it. Check the errors above and the operator-rendered config; this pod likely needs manual investigation or a new rollout." + sleep 30 & + wait $! + done + fi + "#, + bootstrap_servers = bootstrap_servers, + metrics_port = METRICS_PORT, + binary = KAFKA_METADATA_QUORUM_BINARY, + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, + config_dir = STACKABLE_CONFIG_DIR, + controller_properties_file = ConfigFileName::ControllerProperties, + admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH, + add_controller_config = ADD_CONTROLLER_PROPERTIES_PATH, + cli_timeout = CLI_CALL_TIMEOUT_SECONDS, + ) +} + +/// The sidecar's `preStop` command: before this controller pod terminates, check that +/// removing it would not remove the *last* remaining voter from the quorum, and if so, +/// remove it from the voter set. Always exits 0 — a stuck or failed check must never block +/// pod termination. +/// +/// Removing a departing voter only ever *lowers* the majority threshold for the remaining +/// set, and the `remove-controller` RPC itself needs the *current* quorum to already commit +/// it — if peers are unreachable the call simply fails, it can't corrupt anything. So the +/// only real invariant worth enforcing here is "never remove the last voter": a 1-voter +/// quorum can't be reduced further without permanently losing all fault tolerance (there +/// would be no other voter left to ever add a replacement to). +/// +/// This controller's own KRaft node id is derived at runtime from `$POD_NAME` and +/// `$NODE_ID_OFFSET` ([`DERIVE_POD_INDEX`]/[`EXPORT_REPLICA_ID`]), exactly as the `kafka` +/// container's own entrypoint does — see `controller_kafka_container_command`. +/// +/// `describe --replication`'s column layout (`NodeId` as column 1, `DirectoryId` as column +/// 2, `Status` as the last column, with `Status` one of `Leader`/`Follower`/`Observer`) is +/// the *documented* KIP-853 tabular format, but has not been confirmed against a live +/// cluster (see Task 4's brief, Step 5 — deferred to Task 7's kuttl run, which has one). +/// Filtering is deliberately conservative: only rows whose `Status` is a recognized voter +/// value (`Leader`/`Follower`) count towards `total_voters`, and if that filter yields zero +/// voters (e.g. because the real column layout differs from what's assumed here), the +/// check simply retries rather than treating "no known voters" as "safe to remove" — i.e. +/// this fails closed (skips removal) rather than open on a parsing mismatch. +pub fn quorum_manager_pre_stop_command(bootstrap_servers: &str) -> String { + format!( + r#" + set -uo pipefail + {derive_pod_index} + [ -n "$POD_INDEX" ] || exit 0 + {export_replica_id} + DEADLINE=$((SECONDS + 25)) + while [ "$SECONDS" -lt "$DEADLINE" ]; do + describe=$(timeout {cli_timeout} {binary} --bootstrap-controller '{bootstrap_servers}' --command-config {config} describe --replication 2>/dev/null) + if [ -n "$describe" ]; then + voters=$(echo "$describe" | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"') + total_voters=$(echo "$voters" | grep -c .) + if [ "$total_voters" -gt 0 ]; then + remaining_after_removal=$(( total_voters - 1 )) + if [ "$remaining_after_removal" -ge 1 ]; then + directory_id=$(echo "$voters" | awk -v id="$REPLICA_ID" '$1 == id {{ print $2 }}') + if [ -n "$directory_id" ]; then + echo "Removing self (node $REPLICA_ID, directory $directory_id) from the voter set..." + timeout {cli_timeout} {binary} --bootstrap-controller '{bootstrap_servers}' --command-config {config} remove-controller \ + --controller-id "$REPLICA_ID" --controller-directory-id "$directory_id" \ + || echo "remove-controller failed, proceeding with termination anyway" + else + echo "Could not find own node $REPLICA_ID among current voters (already removed?), nothing to do" + fi + break + else + echo "Removing self would leave zero voters, skipping and retrying..." + fi + else + echo "Could not identify any voters in the describe output (unrecognized format), skipping removal for safety and retrying..." + fi + fi + sleep 2 + done + exit 0 + "#, + bootstrap_servers = bootstrap_servers, + binary = KAFKA_METADATA_QUORUM_BINARY, + config = ADMIN_CLIENT_PROPERTIES_PATH, + cli_timeout = CLI_CALL_TIMEOUT_SECONDS, + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, + ) +} + fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor]) -> String { controller_descriptors .iter() @@ -195,15 +410,181 @@ fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor]) -> Stri .join(",") } -fn initial_controllers_command( - controller_descriptors: &[KafkaPodDescriptor], - product_version: &str, -) -> String { - match product_version.starts_with("3.7") { - true => "".to_string(), - false => format!( - "--initial-controllers {initial_controllers}", - initial_controllers = to_initial_controllers(controller_descriptors), - ), +fn initial_controllers_command(controller_descriptors: &[KafkaPodDescriptor]) -> String { + format!( + "--initial-controllers {initial_controllers}", + initial_controllers = to_initial_controllers(controller_descriptors), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + assert!(command.contains("--bootstrap-controller 'controller-0:9093,controller-1:9093'")); + assert!(command.contains("add-controller")); + assert!(!command.contains("--bootstrap-controller 'localhost")); + } + + /// Checks only that the trap and the interruptible-sleep pair are present in the + /// generated command *string* — it does not execute the script, so it cannot verify the + /// trap actually fires promptly under a real `SIGTERM`. That was confirmed separately on + /// a live cluster: without a `TERM` trap, this loop runs as the container's PID 1, whose + /// unhandled signals the kernel suppresses by default — so the `kafka` container in the + /// same pod shut down promptly on `SIGTERM` while this sidecar kept looping (curl + /// connection-refused every ~15s) until Kubernetes gave up and sent `SIGKILL` after the + /// full `terminationGracePeriodSeconds` (1800s), holding the whole pod in `Terminating` + /// well past kuttl's step timeout. The trap plus `sleep 10 &` / `wait $!` (rather than a + /// foreground `sleep 10`) let bash notice and act on `SIGTERM` immediately instead of + /// only after the next blocking command returns. + #[test] + fn quorum_manager_container_command_traps_term_and_sleeps_interruptibly() { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + assert!(command.contains("trap 'exit 0' TERM")); + assert!(command.contains("sleep 10 &")); + assert!(command.contains("wait $!")); + } + + /// Checks only that the generated command *string* concatenates the two config files in + /// the order that makes `add-controller` self-register successfully — it does not + /// execute the script, so it cannot verify runtime behavior. That was confirmed + /// separately on a live cluster: `add-controller` reads `node.id` and its own + /// `listeners`/`controller.listener.names` from the *same* `--command-config` file it + /// connects with, to build the voter registration payload — pointed at the plain + /// admin-client config (which has no `node.id`), every attempt failed with `node.id not + /// found in configuration file`, so no controller was ever admitted as a voter. See + /// [`ADD_CONTROLLER_PROPERTIES_PATH`] for why the fix is a merged file (in this specific + /// order) rather than switching to `controller.properties` outright (that file has no + /// bare `ssl.*`/`security.protocol`, so the AdminClient couldn't reach the TLS-only + /// bootstrap controller at all). + #[test] + fn quorum_manager_container_command_string_merges_controller_and_admin_client_properties_for_add_controller() + { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + // Renders this controller's own `controller.properties` (carries `node.id` and + // `listeners`) via the same REPLICA_ID derivation used by the `kafka` container. + assert!(command.contains("export REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET))")); + assert!(command.contains("config-utils template /tmp/controller.properties")); + // Merges it with the plain admin-client config (carries `security.protocol`/`ssl.*`), + // controller.properties first so the client TLS config in admin-client.properties + // wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`'s doc comment). + assert!(command.contains( + "cat /tmp/controller.properties /stackable/config/admin-client.properties > /tmp/add-controller.properties" + )); + assert!(command.contains("--command-config /tmp/add-controller.properties add-controller")); + } + + #[test] + fn quorum_manager_pre_stop_command_always_exits_zero() { + let command = quorum_manager_pre_stop_command("controller-0:9093,controller-1:9093"); + assert!(command.trim_end().ends_with("exit 0")); + assert!(command.contains("remove-controller")); + } + + /// The old majority-based guard (`majority=$(( total_voters / 2 + 1 ))`, + /// `remaining_after_removal -ge majority`) always blocked the last safe removal of a + /// 2-voter quorum (2 -> 1): `majority` was 2, `remaining_after_removal` was 1, and + /// `1 -ge 2` is false. That left a 2-voter quorum with only 1 live member — a dead + /// quorum requiring manual recovery, exactly the outage this feature exists to prevent. + /// The only invariant that actually matters is "never remove the last voter", so this + /// asserts the generated script uses that condition instead. + #[test] + fn quorum_manager_pre_stop_command_allows_removing_the_second_to_last_voter() { + let command = quorum_manager_pre_stop_command("controller-0:9093,controller-1:9093"); + assert!( + command.contains(r#"remaining_after_removal" -ge 1 ]"#), + "expected the guard to allow removal whenever at least one voter remains \ + afterwards, command was: {command}" + ); + assert!( + !command.contains("majority"), + "the old majority-based guard variable should be gone entirely, command was: \ + {command}" + ); + } + + /// Directly exercises the corrected guard's arithmetic (mirrored from the generated + /// script) end to end in bash: a 2-voter quorum must allow removing the departing voter + /// (leaving 1), while a 1-voter quorum must not (that would leave zero). + #[test] + fn quorum_manager_pre_stop_guard_arithmetic_allows_two_to_one_but_not_one_to_zero() { + fn removal_allowed(total_voters: u32) -> bool { + let script = format!( + r#" + total_voters={total_voters} + remaining_after_removal=$(( total_voters - 1 )) + [ "$remaining_after_removal" -ge 1 ] + "# + ); + std::process::Command::new("bash") + .arg("-c") + .arg(script) + .status() + .expect("bash is available to run this test") + .success() + } + + assert!( + removal_allowed(2), + "removing the second-to-last voter of a 2-voter quorum must be allowed" + ); + assert!( + !removal_allowed(1), + "removing the last voter of a 1-voter quorum must never be allowed" + ); + } + + /// The `preStop` hook already guarded its `REPLICA_ID` derivation against an empty + /// `POD_INDEX`; the main loop's derivation must have the same guard, or an empty + /// `POD_INDEX` would silently produce a wrong `node.id` instead of the sidecar noticing. + #[test] + fn quorum_manager_container_command_guards_against_empty_pod_index() { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + assert!(command.contains(r#"[ -n "$POD_INDEX" ] || exit 0"#)); + } + + /// The render/merge preamble (`cp`/`config-utils template`/`cat`) must log loudly on + /// failure, but must not crash-loop the container: it falls into a degraded loop instead + /// of exiting, and never attempts `add-controller` once degraded. + #[test] + fn quorum_manager_container_command_preamble_is_loud_but_does_not_crash_on_error() { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + // A failed render/merge must not crash-loop the container (that would make the pod + // NotReady and, under OrderedReady pod management, block every sibling pod in the + // role too) — it must log loudly instead and stay Running. + assert!( + !command.contains("set -e"), + "the preamble must not opt into `set -e` (that would crash-loop the container), \ + command was: {command}" + ); + assert!( + command.contains("ERROR"), + "expected a clear error message on a failed render/merge, command was: {command}" + ); + // On failure it must degrade into a loop rather than exiting (which would also crash + // the container) and must never attempt add-controller once degraded. + let error_branch_start = command + .find("echo \"ERROR: quorum-manager failed to render or merge") + .expect("the command has a degraded-mode error branch"); + let degraded_branch = &command[error_branch_start..]; + assert!(degraded_branch.contains("while true")); + // The degraded branch must never invoke the CLI tool (there is no valid rendered + // config to use) — check for the actual invocation, not just the word + // "add-controller" (which also appears inside the degraded branch's own log + // message, explaining what it is *not* doing). + assert!(!degraded_branch.contains(KAFKA_METADATA_QUORUM_BINARY)); + } + + /// The SIGTERM-handling fix's whole point is prompt shutdown, but an unresponsive (not + /// refused) connection to the metrics port would otherwise block the loop body + /// indefinitely — the trap can only fire between commands or during `wait` — reintroducing + /// the exact stall the fix targeted. + #[test] + fn quorum_manager_container_command_metrics_curl_has_timeouts() { + let command = quorum_manager_container_command("controller-0:9093,controller-1:9093"); + assert!(command.contains("curl -s --max-time 5 --connect-timeout 2 localhost")); } } diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 9bf436bb..e26a1277 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -163,13 +163,17 @@ pub fn build(cluster: &ValidatedCluster) -> Result mod tests { use stackable_operator::kube::Resource; - use super::build; - use crate::controller::{ - ValidatedCluster, - test_support::{ - bootstrap_listener, ingress_address, minimal_kafka, validated_cluster, - zookeeper_mode_cluster, + use super::{build, security::STACKABLE_TLS_KAFKA_INTERNAL_DIR}; + use crate::{ + controller::{ + ValidatedCluster, + node_id_hasher::node_id_hash32_offset, + test_support::{ + bootstrap_listener, ingress_address, minimal_kafka, validated_cluster, + zookeeper_mode_cluster, + }, }, + crd::{STACKABLE_CONFIG_DIR, role::KafkaRole}, }; /// Sorted `metadata.name`s of the given resources, for order-independent assertions. @@ -299,6 +303,91 @@ mod tests { ); } + /// The `quorum-manager` sidecar's admin-client calls need every directory that + /// `controller_admin_client_properties` (see `build/security.rs`) writes paths into: + /// the config volume (for `admin-client.properties` itself) and the internal TLS + /// volume (for the keystore/truststore the properties file points at). Missing either + /// mount makes every `add-controller`/`remove-controller` invocation fail SSL init. + #[test] + fn quorum_manager_sidecar_mounts_every_directory_referenced_by_admin_client_properties() { + let cluster = kraft_mode_cluster(); + let resources = build(&cluster).expect("build succeeds"); + + let controller_sts = resources + .stateful_sets + .iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller StatefulSet should be built"); + let pod_spec = controller_sts + .spec + .as_ref() + .expect("the StatefulSet should have a spec") + .template + .spec + .as_ref() + .expect("the pod template should have a spec"); + let quorum_manager = pod_spec + .containers + .iter() + .find(|c| c.name == "quorum-manager") + .expect("the controller pod should have a quorum-manager sidecar"); + + let mount_paths: Vec<&str> = quorum_manager + .volume_mounts + .as_ref() + .expect("the sidecar should have volume mounts") + .iter() + .map(|vm| vm.mount_path.as_str()) + .collect(); + assert!( + mount_paths.contains(&STACKABLE_CONFIG_DIR), + "the sidecar must mount the config directory carrying admin-client.properties, got: {mount_paths:?}" + ); + assert!( + mount_paths.contains(&STACKABLE_TLS_KAFKA_INTERNAL_DIR), + "the sidecar must mount the internal TLS directory admin-client.properties points its keystore/truststore at, got: {mount_paths:?}" + ); + } + + /// Guards against `add_common_kafka_env`'s refactor (accepting a pre-computed + /// `node_id_offset: &str` instead of computing it internally) silently changing the + /// broker's own `NODE_ID_OFFSET` env var value. + #[test] + fn broker_node_id_offset_env_var_is_unchanged_by_the_shared_computation_refactor() { + let cluster = kraft_mode_cluster(); + let resources = build(&cluster).expect("build succeeds"); + + let broker_sts = resources + .stateful_sets + .iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-broker-default")) + .expect("the broker StatefulSet should be built"); + let kafka_container = broker_sts + .spec + .as_ref() + .expect("the StatefulSet should have a spec") + .template + .spec + .as_ref() + .expect("the pod template should have a spec") + .containers + .iter() + .find(|c| c.name == "kafka") + .expect("the broker pod should have a kafka container"); + + let node_id_offset_value = kafka_container + .env + .as_ref() + .expect("the kafka container should have env vars") + .iter() + .find(|env_var| env_var.name == "NODE_ID_OFFSET") + .and_then(|env_var| env_var.value.as_deref()) + .expect("NODE_ID_OFFSET should be set"); + + let expected = node_id_hash32_offset(&KafkaRole::Broker, "default").to_string(); + assert_eq!(node_id_offset_value, expected); + } + /// ZooKeeper mode has no `controller` role, so `build()` emits no controller resources while /// still producing the broker's bootstrap Listener. #[test] diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs index 2f3951f5..da29680d 100644 --- a/rust/operator-binary/src/controller/build/properties/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/mod.rs @@ -25,6 +25,11 @@ pub enum ConfigFileName { Security, #[strum(serialize = "client.properties")] Client, + /// Client-side (unprefixed `security.protocol`/`ssl.*`) properties for an admin CLI tool + /// (e.g. `kafka-metadata-quorum.sh`) running inside a controller pod. Only written to + /// controller rolegroup `ConfigMap`s. + #[strum(serialize = "admin-client.properties")] + AdminClient, /// JAAS configuration for Kerberos authentication. It has the `.properties` /// extension but is not a Java properties file. #[strum(serialize = "jaas.properties")] @@ -86,8 +91,55 @@ mod tests { ); assert_eq!(ConfigFileName::Security.to_string(), "security.properties"); assert_eq!(ConfigFileName::Client.to_string(), "client.properties"); + assert_eq!( + ConfigFileName::AdminClient.to_string(), + "admin-client.properties" + ); assert_eq!(ConfigFileName::Jaas.to_string(), "jaas.properties"); assert_eq!(ConfigFileName::Log4j.to_string(), "log4j.properties"); assert_eq!(ConfigFileName::Log4j2.to_string(), "log4j2.properties"); } + + /// Builds a minimal [`KafkaPodDescriptor`] for the given role, replica and client port. + /// + /// `KafkaPodDescriptor`'s fields are `pub(crate)`, which is crate-wide (not + /// module-scoped) visibility in Rust, so this direct construction is legal from any + /// module inside `stackable-kafka-operator`, including this one. + fn pod_descriptor(role: KafkaRole, replica: u16, client_port: u16) -> KafkaPodDescriptor { + KafkaPodDescriptor { + namespace: "default".parse().expect("valid namespace name"), + role_group_statefulset_name: "kafka-controller-default" + .parse() + .expect("valid statefulset name"), + role_group_service_name: "kafka-controller-default-headless" + .parse() + .expect("valid service name"), + replica, + cluster_domain: stackable_operator::commons::networking::DomainName::try_from( + "cluster.local", + ) + .expect("valid domain"), + node_id: replica.into(), + role, + client_port: client_port.into(), + } + } + + #[test] + fn kraft_controllers_env_value_is_comma_joined_host_ports() { + let pod_descriptors = vec![ + pod_descriptor(KafkaRole::Controller, 0, 9093), + pod_descriptor(KafkaRole::Controller, 1, 9093), + // Brokers must be filtered out of the controller quorum bootstrap servers list. + pod_descriptor(KafkaRole::Broker, 0, 9092), + ]; + + let quorum_bootstrap_servers = kraft_controllers(&pod_descriptors).join(","); + + assert_eq!( + quorum_bootstrap_servers, + "kafka-controller-default-0.kafka-controller-default-headless.default.svc.cluster.local:9093,\ + kafka-controller-default-1.kafka-controller-default-headless.default.svc.cluster.local:9093" + ); + } } 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 284a281b..1869b5e6 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -17,7 +17,7 @@ use crate::{ properties::{ ConfigFileName, config_file_name, product_logging::role_group_config_map_data, }, - security::client_properties, + security::{client_properties, controller_admin_client_properties}, }, }, crd::{ @@ -50,6 +50,16 @@ pub enum Error { role_group: RoleGroupName, }, + #[snafu(display( + "failed to serialize client-side connection properties ([{}] or [{}]) for role group {role_group}", + ConfigFileName::Client, + ConfigFileName::AdminClient + ))] + ClientProperties { + source: PropertiesWriterError, + role_group: RoleGroupName, + }, + #[snafu(display("failed to build pod descriptors"))] BuildPodDescriptors { source: crate::controller::PodDescriptorsError, @@ -159,7 +169,7 @@ pub fn build_rolegroup_config_map( .iter() .filter_map(|(k, v)| v.as_ref().map(|v| (k, v))), ) - .with_context(|_| JvmSecurityPropertiesSnafu { + .with_context(|_| ClientPropertiesSnafu { role_group: role_group_name.clone(), })?, ) @@ -172,6 +182,22 @@ pub fn build_rolegroup_config_map( jaas_config_file(kafka_security.has_kerberos_enabled()), ); + // `admin-client.properties` is only needed by the controller-side sidecar running + // `kafka-metadata-quorum.sh` against the CONTROLLER listener; brokers don't need it. + if let AnyConfig::Controller(_) = &validated_rg.config.config { + cm_builder.add_data( + ConfigFileName::AdminClient.to_string(), + to_java_properties_string( + controller_admin_client_properties(kafka_security) + .iter() + .filter_map(|(k, v)| v.as_ref().map(|v| (k, v))), + ) + .with_context(|_| ClientPropertiesSnafu { + role_group: role_group_name.clone(), + })?, + ); + } + tracing::debug!(?kafka_config, "Applied kafka config"); tracing::debug!(?jvm_sec_props, "Applied JVM config"); diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index c9147c55..737a0f23 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -20,7 +20,7 @@ use stackable_operator::{ apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetUpdateStrategy}, core::v1::{ ConfigMapVolumeSource, ContainerPort, EnvVar, EnvVarSource, ExecAction, - ObjectFieldSelector, PodSpec, Probe, TCPSocketAction, Volume, + LifecycleHandler, ObjectFieldSelector, PodSpec, Probe, TCPSocketAction, Volume, }, }, apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, @@ -49,12 +49,14 @@ use crate::{ build::{ command::{ broker_kafka_container_commands, controller_kafka_container_command, - kafka_log_opts, kafka_log_opts_env_var, + kafka_log_opts, kafka_log_opts_env_var, quorum_manager_container_command, + quorum_manager_pre_stop_command, }, graceful_shutdown::add_graceful_shutdown_config, kerberos::add_kerberos_pod_config, - properties::product_logging::MAX_KAFKA_LOG_FILES_SIZE, + properties::{kraft_controllers, product_logging::MAX_KAFKA_LOG_FILES_SIZE}, security::{ + STACKABLE_TLS_KAFKA_INTERNAL_DIR, STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, add_broker_volume_and_volume_mounts, add_controller_volume_and_volume_mounts, kcat_prober_container_commands, }, @@ -124,7 +126,41 @@ fn common_operator_env_vars( env } +/// Environment variables the operator sets that are common to *every* container in a +/// **controller** pod: today that's the `kafka` server process and, when present, the +/// `quorum-manager` sidecar. +/// +/// The sidecar renders the very same `controller.properties` template (see +/// `properties/controller_properties.rs`) that the `kafka` container's own entrypoint does, to +/// build its own `add-controller`/`remove-controller` config — so it needs every +/// `${env:...}` placeholder that template references (`POD_NAME`, `KAFKA_CLIENT_PORT`, +/// `NAMESPACE`, `ROLEGROUP_HEADLESS_SERVICE_NAME`, `CLUSTER_DOMAIN`). Building this set once +/// and handing it to both containers means they can't silently drift apart over time (a real +/// bug found in review: the sidecar was originally given only `POD_NAME`/`NODE_ID_OFFSET`, +/// so its `controller.properties` render most likely produced a broken `listeners` value). +/// +/// The caller merges the user's `envOverrides` on top (so a user override wins on a name +/// collision) and, for the `kafka` container only, adds container-specific env vars such as +/// `PRE_STOP_CONTROLLER_SLEEP_SECONDS`. +fn controller_pod_shared_env_vars( + validated_cluster: &ValidatedCluster, + kafka_security: &ValidatedKafkaSecurity, + resource_names: &ResourceNames, +) -> EnvVarSet { + common_operator_env_vars(validated_cluster, kafka_security) + .with_field_path(&env_var_name("NAMESPACE"), &FieldPathEnvVar::Namespace) + .with_value( + &env_var_name("ROLEGROUP_HEADLESS_SERVICE_NAME"), + resource_names.headless_service_name().to_string(), + ) + .with_value( + &env_var_name("CLUSTER_DOMAIN"), + validated_cluster.cluster_domain.to_string(), + ) +} + const POD_MANAGEMENT_POLICY_PARALLEL: &str = "Parallel"; +const POD_MANAGEMENT_POLICY_ORDERED_READY: &str = "OrderedReady"; #[derive(Snafu, Debug)] pub enum Error { @@ -271,9 +307,10 @@ pub fn build_broker_rolegroup_statefulset( .pod_descriptors(Some(&KafkaRole::Controller)) .context(BuildPodDescriptorsSnafu)?, kafka_security, - &resolved_product_image.product_version, )]); + let node_id_offset = node_id_hash32_offset(kafka_role, role_group_name.as_ref()).to_string(); + add_common_kafka_env( &mut cb_kafka, merged_config, @@ -281,8 +318,7 @@ pub fn build_broker_rolegroup_statefulset( .product_specific_common_config .jvm_argument_overrides, resolved_product_image, - kafka_role, - role_group_name, + &node_id_offset, )?; cb_kafka @@ -472,22 +508,34 @@ pub fn build_controller_rolegroup_statefulset( let mut pod_builder = PodBuilder::new(); + let node_id_offset = node_id_hash32_offset(kafka_role, role_group_name.as_ref()).to_string(); + // Operator-set env vars first (common + controller-specific); the user's `envOverrides` - // are merged on top and win. - let env: Vec = common_operator_env_vars(validated_cluster, kafka_security) - .with_field_path(&env_var_name("NAMESPACE"), &FieldPathEnvVar::Namespace) - .with_value( - &env_var_name("ROLEGROUP_HEADLESS_SERVICE_NAME"), - resource_names.headless_service_name().to_string(), - ) - .with_value( - &env_var_name("CLUSTER_DOMAIN"), - validated_cluster.cluster_domain.to_string(), - ) + // are merged on top and win. Shared between the `kafka` container and the + // `quorum-manager` sidecar (see `controller_pod_shared_env_vars`) so they can't drift + // apart; each container then layers its own additions on top. + let controller_shared_env = + controller_pod_shared_env_vars(validated_cluster, kafka_security, &resource_names); + + let env: Vec = controller_shared_env + .clone() .with_value(&env_var_name("PRE_STOP_CONTROLLER_SLEEP_SECONDS"), "10") .merge(validated_rg.env_overrides.clone()) .into(); + let quorum_manager_env: Vec = controller_shared_env + .with_value(&env_var_name(KAFKA_NODE_ID_OFFSET), &node_id_offset) + .merge(validated_rg.env_overrides.clone()) + .into(); + + let controller_pod_descriptors = validated_cluster + .pod_descriptors(Some(kafka_role)) + .context(BuildPodDescriptorsSnafu)?; + // Comma-joined `host:port` list of all KRaft controller voters, consumed by the + // `quorum-manager` sidecar container so it can talk to the quorum via + // `kafka-metadata-quorum.sh`. + let quorum_bootstrap_servers = kraft_controllers(&controller_pod_descriptors).join(","); + cb_kafka .image_from_product_image(resolved_product_image) .command(vec![ @@ -498,10 +546,7 @@ pub fn build_controller_rolegroup_statefulset( "-c".to_string(), ]) .args(vec![controller_kafka_container_command( - validated_cluster - .pod_descriptors(Some(kafka_role)) - .context(BuildPodDescriptorsSnafu)?, - &resolved_product_image.product_version, + controller_pod_descriptors, )]); add_common_kafka_env( @@ -511,8 +556,7 @@ pub fn build_controller_rolegroup_statefulset( .product_specific_common_config .jvm_argument_overrides, resolved_product_image, - kafka_role, - role_group_name, + &node_id_offset, )?; cb_kafka @@ -580,6 +624,15 @@ pub fn build_controller_rolegroup_statefulset( .add_container(kafka_container) .affinity(&merged_config.affinity); + if let Some(quorum_manager_container) = build_quorum_manager_container( + resolved_product_image, + kafka_security, + &quorum_bootstrap_servers, + quorum_manager_env, + )? { + pod_builder.add_container(quorum_manager_container); + } + add_common_pod_config( &mut pod_builder, &resource_names, @@ -617,7 +670,7 @@ pub fn build_controller_rolegroup_statefulset( .with_label(RESTART_CONTROLLER_ENABLED_LABEL.to_owned()) .build(), spec: Some(StatefulSetSpec { - pod_management_policy: Some(POD_MANAGEMENT_POLICY_PARALLEL.to_string()), + pod_management_policy: Some(POD_MANAGEMENT_POLICY_ORDERED_READY.to_string()), update_strategy: Some(StatefulSetUpdateStrategy { type_: Some("RollingUpdate".to_string()), ..StatefulSetUpdateStrategy::default() @@ -669,13 +722,17 @@ fn container_ports(kafka_security: &ValidatedKafkaSecurity) -> Vec Result<(), Error> { cb_kafka .add_env_var( @@ -703,10 +760,7 @@ fn add_common_kafka_env( "CONTAINERDEBUG_LOG_DIRECTORY", format!("{STACKABLE_LOG_DIR}/containerdebug"), ) - .add_env_var( - KAFKA_NODE_ID_OFFSET, - node_id_hash32_offset(kafka_role, role_group_name.as_ref()).to_string(), - ); + .add_env_var(KAFKA_NODE_ID_OFFSET, node_id_offset); Ok(()) } @@ -781,6 +835,79 @@ fn container_name(container: impl std::fmt::Display) -> ContainerName { .expect("a container enum variant is always a valid ContainerName") } +/// Name of the controller's `quorum-manager` sidecar container. +const QUORUM_MANAGER_CONTAINER_NAME: &str = "quorum-manager"; + +/// Builds the `quorum-manager` sidecar for a controller pod. Returns `None` when Kerberos is +/// enabled (the sidecar's admin-client properties file only covers the TLS/SSL case). +/// +/// `env` is expected to be [`controller_pod_shared_env_vars`] (plus `NODE_ID_OFFSET` and the +/// rolegroup's `envOverrides`) — the same base the `kafka` container in this pod gets — so +/// this sidecar's `controller.properties` render has every env var it references. See +/// [`controller_pod_shared_env_vars`] for why that matters. +fn build_quorum_manager_container( + resolved_product_image: &ResolvedProductImage, + kafka_security: &ValidatedKafkaSecurity, + quorum_bootstrap_servers: &str, + env: Vec, +) -> Result, Error> { + if kafka_security.has_kerberos_enabled() { + return Ok(None); + } + + let mut cb = ContainerBuilder::new(QUORUM_MANAGER_CONTAINER_NAME).context( + InvalidContainerNameSnafu { + name: QUORUM_MANAGER_CONTAINER_NAME, + }, + )?; + + cb.image_from_product_image(resolved_product_image) + .command(vec![ + "/bin/bash".to_string(), + "-c".to_string(), + quorum_manager_container_command(quorum_bootstrap_servers), + ]) + .add_env_vars(env) + // `kafka-metadata-quorum.sh` goes through `kafka-run-class.sh`, which defaults + // `KAFKA_HEAP_OPTS` to `-Xmx256M` when unset. Set an explicit, modest heap so the + // JVM's max heap plus its base/metaspace/SSL-buffer overhead stays comfortably + // under the container's memory limit below. + .add_env_var(KAFKA_HEAP_OPTS, "-Xmx128M") + .resources( + ResourceRequirementsBuilder::new() + .with_cpu_request("100m") + // A JVM cold start plus an SSL handshake and an admin-client round-trip all + // need to happen inside this sidecar's existing `timeout 15`/`25s preStop` + // budgets (see `CLI_CALL_TIMEOUT_SECONDS` in `command.rs`). + .with_cpu_limit("500m") + .with_memory_request("256Mi") + .with_memory_limit("512Mi") + .build(), + ) + .add_volume_mount(STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) + .context(AddVolumeMountSnafu)? + // `controller_admin_client_properties` (see `build/security.rs`) always points + // its keystore/truststore at this directory, so the sidecar's admin-client calls + // need it mounted here too, not just on the `kafka` container. + .add_volume_mount( + STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, + STACKABLE_TLS_KAFKA_INTERNAL_DIR, + ) + .context(AddVolumeMountSnafu)? + .lifecycle_pre_stop(LifecycleHandler { + exec: Some(ExecAction { + command: Some(vec![ + "/bin/bash".to_string(), + "-c".to_string(), + quorum_manager_pre_stop_command(quorum_bootstrap_servers), + ]), + }), + ..LifecycleHandler::default() + }); + + Ok(Some(cb.build())) +} + fn add_vector_container( pod_builder: &mut PodBuilder, vector_container_name: &ContainerName, @@ -801,3 +928,285 @@ fn add_vector_container( )); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::controller::test_support::{minimal_kafka, validated_cluster}; + + fn kraft_mode_cluster() -> crate::controller::ValidatedCluster { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + metadataManager: kraft + controllers: + roleGroups: + default: + replicas: 3 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + validated_cluster(&kafka) + } + + #[test] + fn controller_statefulset_uses_ordered_ready_pod_management() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller StatefulSet is built"); + + assert_eq!( + sts.spec + .expect("the StatefulSet has a spec") + .pod_management_policy, + Some("OrderedReady".to_string()) + ); + } + + #[test] + fn broker_statefulset_still_uses_parallel_pod_management() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-broker-default")) + .expect("the broker StatefulSet is built"); + + assert_eq!( + sts.spec + .expect("the StatefulSet has a spec") + .pod_management_policy, + Some("Parallel".to_string()) + ); + } + + fn controller_containers( + cluster: &crate::controller::ValidatedCluster, + ) -> Vec { + let resources = crate::controller::build::build(cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller StatefulSet is built"); + sts.spec + .expect("the StatefulSet has a spec") + .template + .spec + .expect("the pod template has a spec") + .containers + } + + #[test] + fn controller_pods_get_a_quorum_manager_sidecar_on_supported_versions() { + let cluster = kraft_mode_cluster(); + let containers = controller_containers(&cluster); + + assert!( + containers + .iter() + .any(|c| c.name == QUORUM_MANAGER_CONTAINER_NAME), + "expected a quorum-manager sidecar, got containers: {:?}", + containers.iter().map(|c| &c.name).collect::>() + ); + } + + #[test] + fn quorum_manager_sidecar_targets_bootstrap_servers_in_its_command() { + let cluster = kraft_mode_cluster(); + let containers = controller_containers(&cluster); + let sidecar = containers + .iter() + .find(|c| c.name == QUORUM_MANAGER_CONTAINER_NAME) + .expect("the quorum-manager sidecar is built"); + + let command = sidecar + .command + .as_ref() + .expect("the sidecar has a command") + .join(" "); + assert!(command.contains("add-controller")); + + let pre_stop_command = sidecar + .lifecycle + .as_ref() + .and_then(|l| l.pre_stop.as_ref()) + .and_then(|h| h.exec.as_ref()) + .and_then(|e| e.command.as_ref()) + .expect("the sidecar has a preStop exec hook") + .join(" "); + assert!(pre_stop_command.contains("remove-controller")); + assert!(pre_stop_command.trim_end().ends_with("exit 0")); + } + + /// Every `${env:NAME}` placeholder found in a rendered Java properties (or similar) + /// string, in first-seen order, de-duplicated. + /// + /// The Java properties writer used to serialize the rendered `controller.properties` + /// escapes `:` as `\:` (`:` otherwise separates a properties key from its value), so a + /// placeholder actually appears as `${env\:NAME}` in the rendered ConfigMap content — + /// this accepts either form. + fn extract_env_placeholders(rendered: &str) -> Vec { + let mut result = Vec::new(); + let mut rest = rendered; + while let Some(start) = rest.find("${env") { + rest = &rest[start + "${env".len()..]; + rest = rest.strip_prefix('\\').unwrap_or(rest); + let Some(rest_after_colon) = rest.strip_prefix(':') else { + continue; + }; + rest = rest_after_colon; + let Some(end) = rest.find('}') else { + break; + }; + let name = rest[..end].to_string(); + if !result.contains(&name) { + result.push(name); + } + rest = &rest[end + 1..]; + } + result + } + + /// Regression test for a real bug found in review: `build_quorum_manager_container` once + /// set only `POD_NAME`/`NODE_ID_OFFSET` on the sidecar, while its own + /// `controller.properties` render (used to build the `add-controller` config, see + /// `command.rs`) needs `POD_NAME`, `ROLEGROUP_HEADLESS_SERVICE_NAME`, `NAMESPACE`, + /// `CLUSTER_DOMAIN` and `KAFKA_CLIENT_PORT` — so the rendered `listeners` value was most + /// likely broken (unresolved `${env:...}` placeholders). This asserts, from the actual + /// rendered `controller.properties` content, that every placeholder it references has a + /// matching env var on the sidecar container. + #[test] + fn quorum_manager_sidecar_has_every_env_var_controller_properties_rendering_references() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + + let controller_properties = resources + .config_maps + .iter() + .find(|cm| cm.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller rolegroup ConfigMap is built") + .data + .as_ref() + .expect("the ConfigMap carries data") + .get("controller.properties") + .expect("controller.properties is rendered into the ConfigMap") + .clone(); + + let placeholders = extract_env_placeholders(&controller_properties); + assert!( + placeholders.len() > 1, + "sanity check failed: expected multiple ${{env:...}} placeholders in the rendered \ + controller.properties, got: {placeholders:?}" + ); + + let containers = controller_containers(&cluster); + let sidecar = containers + .iter() + .find(|c| c.name == QUORUM_MANAGER_CONTAINER_NAME) + .expect("the quorum-manager sidecar is built"); + let sidecar_env_names: Vec<&str> = sidecar + .env + .as_ref() + .expect("the sidecar has env vars") + .iter() + .map(|e| e.name.as_str()) + .collect(); + + for placeholder in &placeholders { + // REPLICA_ID is not a Kubernetes-injected env var: both the `kafka` container's + // entrypoint and this sidecar's main-loop script derive and `export` it + // themselves from `$POD_NAME`/`$NODE_ID_OFFSET` before rendering the template + // (see `command.rs`), so it's expected to be absent from the container spec's + // `env` list. + if placeholder == "REPLICA_ID" { + continue; + } + assert!( + sidecar_env_names.contains(&placeholder.as_str()), + "quorum-manager sidecar is missing env var {placeholder:?}, which is \ + referenced by controller.properties's rendering; sidecar env vars: \ + {sidecar_env_names:?}" + ); + } + + // Targeted assertion (rather than relying on it only showing up incidentally among + // `placeholders` above): NODE_ID_OFFSET is consumed directly by the sidecar's own + // `EXPORT_REPLICA_ID` bash logic under `set -u` (see `command.rs`), so a regression + // here would break the sidecar's main loop and its `preStop` hook silently (an unset + // variable under `set -u` aborts the script). + assert!( + sidecar_env_names.contains(&KAFKA_NODE_ID_OFFSET), + "quorum-manager sidecar is missing the {KAFKA_NODE_ID_OFFSET} env var, needed by \ + its EXPORT_REPLICA_ID derivation under `set -u`; sidecar env vars: \ + {sidecar_env_names:?}" + ); + } + + #[test] + fn controller_pods_get_no_quorum_manager_sidecar_when_kerberos_is_enabled() { + // This is a Global Constraint (see the plan header): the sidecar's admin-client + // properties file only covers the TLS/SSL case, so it must never be added when + // Kerberos is enabled, even on an otherwise-supported Kafka version. + // + // Rather than building a full CRD-level Kerberos fixture (which needs a resolved + // AuthenticationClass threaded through `DereferencedObjects`, more than this test + // needs), call `build_quorum_manager_container` directly — it already takes + // `&ValidatedKafkaSecurity` as a parameter, so a fixture at that level is enough. + // Reuse the `kerberos()` fixture from `security.rs`'s existing test module (see + // Task 2). + let cluster = kraft_mode_cluster(); + let kerberos_security = crate::controller::build::security::tests::kerberos(); + + let result = build_quorum_manager_container( + &cluster.image, + &kerberos_security, + "controller-0:9093", + Vec::new(), + ) + .expect("build_quorum_manager_container does not error for a kerberos security value"); + + assert!(result.is_none()); + } + + #[test] + fn broker_pods_never_get_a_quorum_manager_sidecar() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + let sts = resources + .stateful_sets + .into_iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-broker-default")) + .expect("the broker StatefulSet is built"); + let containers = sts + .spec + .expect("the StatefulSet has a spec") + .template + .spec + .expect("the pod template has a spec") + .containers; + + assert!( + !containers + .iter() + .any(|c| c.name == QUORUM_MANAGER_CONTAINER_NAME) + ); + } +} diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index e36191c1..87693836 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -49,8 +49,11 @@ const PROPERTY_SECURITY_PROTOCOL: &str = "security.protocol"; const PROPERTY_SASL_ENABLED_MECHANISMS: &str = "sasl.enabled.mechanisms"; const PROPERTY_SASL_KERBEROS_SERVICE_NAME: &str = "sasl.kerberos.service.name"; const PROPERTY_SASL_INTER_BROKER_MECHANISM: &str = "sasl.mechanism.inter.broker.protocol"; -const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &str = "/stackable/tls-kafka-internal"; -const STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: &str = "tls-kafka-internal"; +// Also mounted on the controller's `quorum-manager` sidecar (see +// `build_quorum_manager_container` in `build/resource/statefulset.rs`), since +// `controller_admin_client_properties` points its keystore/truststore here. +pub(crate) const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &str = "/stackable/tls-kafka-internal"; +pub(crate) const STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: &str = "tls-kafka-internal"; const STACKABLE_TLS_KAFKA_SERVER_DIR: &str = "/stackable/tls-kafka-server"; const STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME: &str = "tls-kafka-server"; // directories @@ -221,6 +224,34 @@ pub fn client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Opti props } +/// Client-side (unprefixed `security.protocol`/`ssl.*`) properties for an admin CLI tool +/// (e.g. `kafka-metadata-quorum.sh`) talking to the CONTROLLER listener from *inside* a +/// controller pod, over the `tls-kafka-internal` volume mounted by +/// `add_controller_volume_and_volume_mounts`. +/// +/// This is deliberately separate from `client_properties()`: that function points at +/// `/stackable/tls-kafka-server`, a directory that is only mounted on broker pods. +/// +/// Internal (broker/controller) TLS is mandatory (see [`ValidatedKafkaSecurity::tls_internal_secret_class`]), +/// and `add_controller_volume_and_volume_mounts` unconditionally mounts both the keystore and +/// truststore on controller pods, independent of the external client TLS/authentication +/// settings. So, mirroring how `controller_config_settings` unconditionally writes the +/// CONTROLLER listener's keystore/truststore settings, this function always returns SSL +/// properties - there is no plaintext variant for the CONTROLLER listener. +pub fn controller_admin_client_properties( + _security: &ValidatedKafkaSecurity, +) -> Vec<(String, Option)> { + let mut properties = vec![]; + + properties.push(( + PROPERTY_SECURITY_PROTOCOL.to_string(), + Some(KafkaListenerProtocol::Ssl.to_string()), + )); + push_client_ssl_stores(&mut properties, STACKABLE_TLS_KAFKA_INTERNAL_DIR); + + properties +} + /// Adds required volumes and volume mounts to the broker pod and container builders /// depending on the tls and authentication settings. pub fn add_broker_volume_and_volume_mounts( @@ -651,7 +682,7 @@ fn kcat_client_sasl_ssl(cert_directory: &str, service_name: &str) -> Vec } #[cfg(test)] -mod tests { +pub(crate) mod tests { use std::{collections::BTreeMap, str::FromStr}; use stackable_operator::{ @@ -724,7 +755,7 @@ mod tests { } /// Kerberos, which also requires server and internal TLS. - fn kerberos() -> ValidatedKafkaSecurity { + pub(crate) fn kerberos() -> ValidatedKafkaSecurity { ValidatedKafkaSecurity::new( ResolvedAuthenticationClasses::new(vec![kerberos_auth_class()]), SecretClassName::from_str("tls").expect("tls secret class name is valid"), @@ -888,6 +919,62 @@ mod tests { assert!(props.contains_key("sasl.jaas.config")); } + // ---- controller_admin_client_properties ---- + + #[test] + fn controller_admin_client_properties_uses_the_internal_tls_directory() { + let security = server_tls(); + let props = as_map(controller_admin_client_properties(&security)); + + assert_eq!( + props.get("security.protocol"), + Some(&Some("SSL".to_string())) + ); + assert_eq!( + props.get("ssl.truststore.location"), + Some(&Some( + "/stackable/tls-kafka-internal/truststore.p12".to_string() + )) + ); + assert_eq!( + props.get("ssl.truststore.type"), + Some(&Some("PKCS12".to_string())) + ); + } + + #[test] + fn controller_admin_client_properties_includes_keystore_when_client_auth_is_required() { + let security = client_auth_tls(); + let props = as_map(controller_admin_client_properties(&security)); + + assert_eq!( + props.get("ssl.keystore.location"), + Some(&Some( + "/stackable/tls-kafka-internal/keystore.p12".to_string() + )) + ); + } + + #[test] + fn controller_admin_client_properties_always_uses_tls_even_without_external_client_tls() { + // Internal (broker/controller) TLS is mandatory (`tls_internal_secret_class()` always + // returns a SecretClass, defaulting to "tls"), and `add_controller_volume_and_volume_mounts` + // unconditionally mounts both the keystore and truststore on controller pods, independent + // of the external client TLS/authentication settings. So even the "plaintext" fixture + // (no external client TLS, no client-cert auth) still needs SSL to reach the CONTROLLER + // listener - mirroring `controller_config_settings`'s unconditional treatment of the same + // listener (see `controller_config_plaintext_has_internal_tls`). + let security = plaintext(); + let props = as_map(controller_admin_client_properties(&security)); + + assert_eq!( + props.get("security.protocol"), + Some(&Some("SSL".to_string())) + ); + assert!(props.contains_key("ssl.truststore.location")); + assert!(props.contains_key("ssl.keystore.location")); + } + // ---- broker_config_settings ---- #[test] diff --git a/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 index 61968a8a..08832143 100644 --- a/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 @@ -5,6 +5,17 @@ kind: TestAssert timeout: 600 commands: - script: kubectl -n $NAMESPACE wait --for=condition=available kafkaclusters.kafka.stackable.tech/test-kafka --timeout 301s + - script: | + # :9093 is the TLS client port of this test fixture's default security config, not a + # fixed Kafka port - if the fixture's TLS/port config changes, update this too. + kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ + /stackable/kafka/bin/kafka-metadata-quorum.sh \ + --bootstrap-controller test-kafka-controller-default-0.test-kafka-controller-default-headless.$NAMESPACE.svc.cluster.local:9093 \ + --command-config /stackable/config/admin-client.properties \ + describe --replication | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"' | wc -l | grep -q '^5$' + # `timeout` is known-inert here: kuttl's TestAssert `commands` don't read this field (only + # TestStep commands do); left in place only as documentation of the intended budget. + timeout: 30 --- apiVersion: apps/v1 kind: StatefulSet @@ -13,6 +24,7 @@ metadata: status: readyReplicas: 3 replicas: 3 +--- apiVersion: apps/v1 kind: StatefulSet metadata: diff --git a/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 index cd8c8ae2..f54db021 100644 --- a/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 @@ -5,6 +5,17 @@ kind: TestAssert timeout: 600 commands: - script: kubectl -n $NAMESPACE wait --for=condition=available kafkaclusters.kafka.stackable.tech/test-kafka --timeout 301s + - script: | + # :9093 is the TLS client port of this test fixture's default security config, not a + # fixed Kafka port - if the fixture's TLS/port config changes, update this too. + kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ + /stackable/kafka/bin/kafka-metadata-quorum.sh \ + --bootstrap-controller test-kafka-controller-default-0.test-kafka-controller-default-headless.$NAMESPACE.svc.cluster.local:9093 \ + --command-config /stackable/config/admin-client.properties \ + describe --replication | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"' | wc -l | grep -q '^3$' + # `timeout` is known-inert here: kuttl's TestAssert `commands` don't read this field (only + # TestStep commands do); left in place only as documentation of the intended budget. + timeout: 30 --- apiVersion: apps/v1 kind: StatefulSet @@ -13,6 +24,7 @@ metadata: status: readyReplicas: 3 replicas: 3 +--- apiVersion: apps/v1 kind: StatefulSet metadata: diff --git a/tests/templates/kuttl/operations-kraft/README.md b/tests/templates/kuttl/operations-kraft/README.md index 5c0fa86b..f838cef7 100644 --- a/tests/templates/kuttl/operations-kraft/README.md +++ b/tests/templates/kuttl/operations-kraft/README.md @@ -12,3 +12,7 @@ Notes Both brokers and controllers try to communicate with old controllers. This is why, the last step scales from 5 -> 3 controllers. This at least, leaves the cluster in a working state. + This was not re-tested after the `quorum-manager` sidecar was added (see + `docs/modules/kafka/pages/usage-guide/kraft-controller.adoc`); this suite only ever + exercises 3 -> 5 -> 3, so this caveat is left in place until scaling down to 1 is + actually covered by a test.