diff --git a/.gitignore b/.gitignore index 696bc411e..0ff5d1ecd 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 d4d2282a5..a794642f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ All notable changes to this project will be documented in this file. ### Changed +- The dynamic KRaft quorum created by the operator is now scaled automatically. Previously, + manual intervention was needed after every scale operation. + This change introduces a new side-car container (`quorum-manager`) to all controller pods + that adds the new controller to the voter list. + On termination, a new `preStop` hook on the controller container (`kafka`) removes the pod from + the voter list before shutdown. + The property `controller.quorum.bootstrap.servers` now contains the headless service names + of all controller role groups instead of individual peer host names. This prevevents the + restart controller from restarting all pods in the quorum when a new one is added/deleted. + The controller `StatefulSet` is now scaled using `OrderedBy` instead of the `Parallel` strategy + to ensure only one voter is added/removed at a time and thus keep the quorum healthy ([#1010]). - Internal operator refactoring: introduce a build() step in the reconciler that assembles all relevant Kubernetes resources before anything is applied ([#985]). - Bump stackable-operator to 0.116.0 ([#994], [#1011]). @@ -32,11 +43,20 @@ All notable changes to this project will be documented in this file. which could cause problems with GitOps tools (e.g. ArgoCD) reporting a diff in the custom resources. See [our internal issue](https://github.com/stackabletech/hdfs-operator/issues/626) and [the fix](https://github.com/kube-rs/kube/pull/2042) for details ([#998]). +### Removed + +- BREAKING: The broker pod's separate `kcat-prober` sidecar container has been removed; its + `kcat`-based readiness probe now runs directly on the `kafka` container instead (`kcat` has + shipped in the same product image as `kafka` since #527, so the dedicated container/image was + no longer needed). The `kcat-prober` value is no longer accepted in a broker's + `logging.containers` CRD field ([#1010]). + [#985]: https://github.com/stackabletech/kafka-operator/pull/985 [#990]: https://github.com/stackabletech/kafka-operator/pull/990 [#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 +[#1010]: https://github.com/stackabletech/kafka-operator/pull/1010 [#1011]: https://github.com/stackabletech/kafka-operator/pull/1011 ## [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 455188c97..0254bc87a 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -85,13 +85,36 @@ 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. +* Each controller pod runs an additional `quorum-manager` sidecar container that, on startup, admits the pod into + the KRaft voter set (`kafka-metadata-quorum.sh add-controller`). + Removing the pod from the voter set again (`remove-controller`) on termination runs as the `kafka` container's + *own* `preStop` hook. +* Controller pods have a `startupProbe` (a plain TCP check on the KRaft listener port) and a `livenessProbe` that + combines that same TCP check with a check that the controller's local Raft state hasn't been stuck `unattached` + for an extended period — a symptom of a dynamically-joining controller resolving + `controller.quorum.bootstrap.servers` to its own pod address and getting wedged fetching from itself; a restart + forces a fresh DNS resolution attempt. A `readinessProbe` separately checks that the Raft state is one of + `leader`, `follower`, or `voted` via the controller's metrics endpoint, so a controller that cannot join or + rejoin the quorum is correctly reported as not ready instead of appearing healthy. +* Exactly one controller (the one with the numerically lowest `node.id` among all controller pod descriptors) formats + with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself. + Every other controller formats with `--no-initial-controllers` and joins purely through the sidecar's `add-controller` call. + Brokers always format with `--no-initial-controllers` too; they are never voters. + the current replica count. +* `controller.quorum.bootstrap.servers` (used by the `kafka` process itself to find the controller quorum, by the + `quorum-manager` sidecar for its own `add-controller` calls, and by the `kafka` container's own `preStop` hook + for its `remove-controller` call) points at each controller role group's own headless Service DNS name, not + individual pod addresses. == 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. +* The single controller chosen to bootstrap the quorum standalone is picked by a stable, deterministic rule + (lowest `node.id`), which is safe only for a cluster's *original* bootstrap. If that specific controller's + persistent volume is ever lost and needs to reformat after the cluster has already formed a quorum elsewhere, + reformatting it with `--standalone` would bootstrap a second, conflicting one-node quorum instead of rejoining + the existing one. If this happens, recovery requires manual intervention with Kafka's own KRaft tooling. == Troubleshooting @@ -108,10 +131,14 @@ 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 -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). +Controller replicas can be scaled up and down on a running cluster. A per-pod `quorum-manager` sidecar admits the +pod into the KRaft voter set on startup, and the `kafka` container's own `preStop` hook removes it again on +termination, as described under "Internal operator details" above. + +Scaling more than one controller 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. == Kraft migration guide diff --git a/docs/modules/kafka/partials/supported-versions.adoc b/docs/modules/kafka/partials/supported-versions.adoc index 1a57dce0f..5a962cb42 100644 --- a/docs/modules/kafka/partials/supported-versions.adoc +++ b/docs/modules/kafka/partials/supported-versions.adoc @@ -7,10 +7,7 @@ * 3.9.2 (LTS) * 3.9.1 (deprecated) -Support for clusters running in Kraft mode (which includes Apache Kafka 4.x.x) is experimental because it has not been thoroughly tested in production environments yet. +Support for clusters running in Kraft mode (which includes Apache Kafka >= 4.x) is experimental due to the following known issues: -Also there are some known issues such as: - -* Controller scaling is not reliable. * Kerberos authentication is not tested yet. * Service exposition is not definitive. diff --git a/extra/crds.yaml b/extra/crds.yaml index d355b6cf3..910e2698b 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -181,86 +181,6 @@ spec: description: Configuration per logger type: object type: object - kcat-prober: - anyOf: - - required: - - custom - - {} - - {} - description: Log configuration of the container - properties: - console: - description: Configuration for the console appender - nullable: true - properties: - level: - description: |- - The log level threshold. - Log events with a lower log level are discarded. - enum: - - TRACE - - DEBUG - - INFO - - WARN - - ERROR - - FATAL - - NONE - - null - nullable: true - type: string - type: object - custom: - description: Log configuration provided in a ConfigMap - properties: - configMap: - description: ConfigMap containing the log configuration files - nullable: true - type: string - type: object - file: - description: Configuration for the file appender - nullable: true - properties: - level: - description: |- - The log level threshold. - Log events with a lower log level are discarded. - enum: - - TRACE - - DEBUG - - INFO - - WARN - - ERROR - - FATAL - - NONE - - null - nullable: true - type: string - type: object - loggers: - additionalProperties: - description: Configuration of a logger - properties: - level: - description: |- - The log level threshold. - Log events with a lower log level are discarded. - enum: - - TRACE - - DEBUG - - INFO - - WARN - - ERROR - - FATAL - - NONE - - null - nullable: true - type: string - type: object - default: {} - description: Configuration per logger - type: object - type: object vector: anyOf: - required: @@ -717,86 +637,6 @@ spec: description: Configuration per logger type: object type: object - kcat-prober: - anyOf: - - required: - - custom - - {} - - {} - description: Log configuration of the container - properties: - console: - description: Configuration for the console appender - nullable: true - properties: - level: - description: |- - The log level threshold. - Log events with a lower log level are discarded. - enum: - - TRACE - - DEBUG - - INFO - - WARN - - ERROR - - FATAL - - NONE - - null - nullable: true - type: string - type: object - custom: - description: Log configuration provided in a ConfigMap - properties: - configMap: - description: ConfigMap containing the log configuration files - nullable: true - type: string - type: object - file: - description: Configuration for the file appender - nullable: true - properties: - level: - description: |- - The log level threshold. - Log events with a lower log level are discarded. - enum: - - TRACE - - DEBUG - - INFO - - WARN - - ERROR - - FATAL - - NONE - - null - nullable: true - type: string - type: object - loggers: - additionalProperties: - description: Configuration of a logger - properties: - level: - description: |- - The log level threshold. - Log events with a lower log level are discarded. - enum: - - TRACE - - DEBUG - - INFO - - WARN - - ERROR - - FATAL - - NONE - - null - nullable: true - type: string - type: object - default: {} - description: Configuration per logger - type: object - type: object vector: anyOf: - required: diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 15b53fa60..a12a3cd01 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -607,6 +607,14 @@ pub(crate) mod test_support { /// Runs the real validate step against a minimal (auth/OPA-free) fixture. pub fn validated_cluster(kafka: &v1alpha1::KafkaCluster) -> ValidatedCluster { + validate_err(kafka).expect("validate should succeed for the test fixture") + } + + /// Runs the real validate step against a minimal (auth/OPA-free) fixture, without unwrapping + /// the result -- for tests asserting on a specific validation failure. + pub fn validate_err( + kafka: &v1alpha1::KafkaCluster, + ) -> Result { validate( kafka, DereferencedObjects { @@ -617,7 +625,6 @@ pub(crate) mod test_support { }, &operator_environment(), ) - .expect("validate should succeed for the test fixture") } } diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 34e303621..e9cd729a0 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -6,6 +6,7 @@ use stackable_operator::{ product_logging::framework::{ create_vector_shutdown_file_command, remove_vector_shutdown_file_command, }, + shared::time::Duration, utils::COMMON_BASH_TRAP_FUNCTIONS, v2::{builder::pod::container::EnvVarName, product_logging::framework::STACKABLE_LOG_DIR}, }; @@ -14,8 +15,8 @@ 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, - STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR, + BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, METRICS_PORT, STACKABLE_CONFIG_DIR, + STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR, role::KafkaRole, }, }; @@ -38,12 +39,16 @@ pub fn kafka_log_opts(product_version: &str) -> String { // The env var carrying the Kafka log4j options (see [`kafka_log_opts`]). constant!(pub KAFKA_LOG4J_OPTS: EnvVarName = "KAFKA_LOG4J_OPTS"); +/// 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`). +const DERIVE_POD_INDEX: &str = r#"POD_INDEX=$(echo "$POD_NAME" | grep -oE '[0-9]+$')"#; + +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} @@ -66,18 +71,14 @@ 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), } } -fn broker_start_command( - kraft_mode: bool, - controller_descriptors: Vec, - product_version: &str, -) -> String { +fn broker_start_command(kraft_mode: bool) -> 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\") @@ -89,6 +90,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, @@ -99,11 +102,10 @@ fn broker_start_command( formatdoc! {" {common_command} - bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command} + bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted --no-initial-controllers bin/kafka-server-start.sh /tmp/{properties_file} & ", properties_file = ConfigFileName::BrokerProperties, - initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), } } else { formatdoc! {" @@ -115,104 +117,718 @@ fn broker_start_command( } } -// During a namespace or stacklet delete the Kafka controllers shut down too fast leaving the brokers -// in a bad state. -// Brokers try to connect to controllers before gracefully shutting down but by that time, all -// controllers are already gone. -// The broker pods are then kept alive until the value of `gracefulShutdownTimeout` is reached. -// The environment variable `PRE_STOP_CONTROLLER_SLEEP_SECONDS` delays the termination of the -// controller processes to give the brokers more time to offload data and shutdown gracefully. -// Kubernetes has a built in `pre-stop` hook feature that is not yet generally available on all platforms -// supported by the operator. -const BASH_TRAP_FUNCTIONS: &str = r#" -prepare_signal_handlers() -{ - unset term_child_pid - unset term_kill_needed - trap 'handle_term_signal' TERM -} - -handle_term_signal() -{ - if [ "${term_child_pid}" ]; then - [ -n "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" - kill -TERM "${term_child_pid}" 2>/dev/null - else - term_kill_needed="yes" - fi -} +/// Chooses exactly one controller (the one with the numerically lowest KRaft `node_id` among +/// all controller pod descriptors, a value that is stable across scale-up/down of an existing +/// controller role group, since new replicas only ever get higher node ids) to bootstrap the +/// dynamic KRaft quorum by itself, via `kafka-storage.sh format --standalone`, the first time +/// it is ever formatted. +/// +/// Every other controller — whether it is part of the cluster's initial desired replica count +/// or added later on scale-up — is formatted with `--no-initial-controllers` and relies +/// entirely on the `quorum-manager` sidecar's `add-controller` loop to join the quorum. +/// +/// Known limitation: this rule is only safe for a cluster's *original* bootstrap. If the +/// designated node's persistent volume is ever lost and needs to reformat after the cluster has +/// already formed a quorum elsewhere, reformatting it with `--standalone` would bootstrap a +/// second, conflicting one-node quorum instead of rejoining the existing one. +fn controller_quorum_format_flag(controller_descriptors: &[KafkaPodDescriptor]) -> String { + let bootstrap_node_id = controller_descriptors + .iter() + .filter(|descriptor| descriptor.role == KafkaRole::Controller) + .map(|descriptor| descriptor.node_id) + .min() + .unwrap_or(0); -wait_for_termination() -{ - set +e - term_child_pid=$1 - if [[ -v term_kill_needed ]]; then - [ -n "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" - kill -TERM "${term_child_pid}" 2>/dev/null - fi - wait ${term_child_pid} 2>/dev/null - trap - TERM - wait ${term_child_pid} 2>/dev/null - set -e + formatdoc! {" + if [ \"$REPLICA_ID\" = \"{bootstrap_node_id}\" ]; then + FORMAT_QUORUM_FLAG=--standalone + else + FORMAT_QUORUM_FLAG=--no-initial-controllers + fi + " + } } -"#; pub fn controller_kafka_container_command( controller_descriptors: Vec, - product_version: &str, ) -> String { formatdoc! {" - {BASH_TRAP_FUNCTIONS} + {COMMON_BASH_TRAP_FUNCTIONS} {remove_vector_shutdown_file_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} config-utils template /tmp/{properties_file} - bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command} + {quorum_format_flag} + bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted \"$FORMAT_QUORUM_FLAG\" bin/kafka-server-start.sh /tmp/{properties_file} & wait_for_termination $! {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), + quorum_format_flag = controller_quorum_format_flag(&controller_descriptors), create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) } } -fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor]) -> String { - controller_descriptors - .iter() - .map(|desc| desc.as_voter()) - .collect::>() - .join(",") +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. +/// +/// **Order matters.** There is no key overlap between the two files today, but +/// `controller.properties` accepts unconditional `configOverrides` merged into it, +/// so a user override there could add a colliding key. +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`. +const CLI_CALL_TIMEOUT_SECONDS: u32 = 15; + +/// Grace period (seconds) after [`CLI_CALL_TIMEOUT_SECONDS`] elapses before `timeout` sends +/// `SIGKILL`, via `--kill-after`. +const CLI_CALL_KILL_AFTER_SECONDS: u32 = 5; + +/// Shell snippet setting `$BOOTSTRAP_SERVERS` by extracting +/// `controller.quorum.bootstrap.servers` from the static, un-rendered `controller.properties` +/// ConfigMap file. +/// +/// Reading this at runtime, rather than baking the peer list into this script as a Rust +/// literal, keeps both sidecar scripts' content — and therefore the controller pod +/// template — identical across changes to an existing controller role group's *replica +/// count*. +fn extract_bootstrap_servers_command() -> String { + format!( + r#"BOOTSTRAP_SERVERS=$(grep '^controller.quorum.bootstrap.servers=' {config_dir}/{controller_properties_file} | cut -d= -f2- | sed 's/\\:/:/g')"#, + config_dir = STACKABLE_CONFIG_DIR, + controller_properties_file = ConfigFileName::ControllerProperties, + ) +} + +/// 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. +pub fn quorum_manager_container_command() -> String { + format!( + r#" + set -uo pipefail + ADD_CONTROLLER_PID="" + trap 'handle_term_signal' TERM + + handle_term_signal() + {{ + [ -n "$ADD_CONTROLLER_PID" ] && kill -TERM "$ADD_CONTROLLER_PID" 2>/dev/null + exit 0 + }} + + {derive_pod_index} + [ -n "$POD_INDEX" ] || exit 0 + {export_replica_id} + {extract_bootstrap_servers} + + 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 --kill-after={cli_kill_after} {cli_timeout} {binary} --bootstrap-controller "$BOOTSTRAP_SERVERS" --command-config {add_controller_config} add-controller & + ADD_CONTROLLER_PID=$! + wait "$ADD_CONTROLLER_PID" \ + || echo "add-controller attempt failed (this is expected if it already succeeded or a leader election is in progress), will retry" + ADD_CONTROLLER_PID="" + 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 + "#, + metrics_port = METRICS_PORT, + binary = KAFKA_METADATA_QUORUM_BINARY, + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, + extract_bootstrap_servers = extract_bootstrap_servers_command(), + 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, + cli_kill_after = CLI_CALL_KILL_AFTER_SECONDS, + ) } -fn initial_controllers_command( - controller_descriptors: &[KafkaPodDescriptor], - product_version: &str, +/// Reserved (seconds), out of the pod's total `terminationGracePeriodSeconds`, for the `kafka` +/// process's *own* `SIGTERM`-triggered shutdown after this preStop hook finishes or gives up. +const PRE_STOP_RESERVED_FOR_KAFKA_SHUTDOWN_SECONDS: u64 = 30; + +/// Floor for [`pre_stop_deadline_seconds`]: never worse than the original fixed budget, even +/// for a user-configured `gracefulShutdownTimeout` too short to leave +/// [`PRE_STOP_RESERVED_FOR_KAFKA_SHUTDOWN_SECONDS`] of headroom. +const PRE_STOP_MIN_DEADLINE_SECONDS: u64 = 25; + +/// Cap for [`pre_stop_deadline_seconds`]: even against a generous `gracefulShutdownTimeout` +/// (the operator's own default is 30 minutes), a single pod's voter removal shouldn't +/// plausibly hang for tens of minutes during a routine scale-down. +const PRE_STOP_MAX_DEADLINE_SECONDS: u64 = 120; + +/// The wall-clock budget (seconds) [`controller_remove_self_pre_stop_command`] retries voter removal +/// for, derived from the pod's actual `gracefulShutdownTimeout` rather than a single fixed +/// constant. +fn pre_stop_deadline_seconds(graceful_shutdown_timeout: Option) -> u64 { + graceful_shutdown_timeout + .map(|timeout| { + timeout + .as_secs() + .saturating_sub(PRE_STOP_RESERVED_FOR_KAFKA_SHUTDOWN_SECONDS) + .clamp(PRE_STOP_MIN_DEADLINE_SECONDS, PRE_STOP_MAX_DEADLINE_SECONDS) + }) + .unwrap_or(PRE_STOP_MIN_DEADLINE_SECONDS) +} + +/// The `kafka` container's own `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. +/// +/// The "would leave zero voters" case is the one exception that does *not* retry: once a +/// `describe` shows this pod is the last remaining voter, stop. +/// +/// IMPORTANT: the last voter must never be removed from the quorum because that would break +/// cluster restarts. In that situation a restart would reformat the Raft metadata effectively +/// losing all information from the previous iteration. +/// +/// If every retry within `DEADLINE` fails, the loop falls through with the voter never +/// actually removed; the final `echo "ERROR: ..."` makes that failure loud (grep/alert-able in +/// container logs) rather than a plain, easy-to-miss log line, since a stale voter entry left +/// behind here is exactly the kind of thing that can strand a later restart-from-zero (see +/// `controller_stuck_unattached_liveness_probe`'s doc comment in `resource/statefulset.rs`). +pub fn controller_remove_self_pre_stop_command( + graceful_shutdown_timeout: Option, ) -> String { - match product_version.starts_with("3.7") { - true => "".to_string(), - false => format!( - "--initial-controllers {initial_controllers}", - initial_controllers = to_initial_controllers(controller_descriptors), - ), - } + format!( + r#" + set -uo pipefail + {derive_pod_index} + [ -n "$POD_INDEX" ] || exit 0 + {export_replica_id} + {extract_bootstrap_servers} + DEADLINE=$((SECONDS + {deadline_seconds})) + finished=false + while [ "$SECONDS" -lt "$DEADLINE" ]; do + describe=$(timeout --kill-after={cli_kill_after} {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..." + if timeout --kill-after={cli_kill_after} {cli_timeout} {binary} --bootstrap-controller "$BOOTSTRAP_SERVERS" --command-config {config} remove-controller \ + --controller-id "$REPLICA_ID" --controller-directory-id "$directory_id"; then + finished=true + else + echo "remove-controller attempt failed, will retry if time remains" + fi + else + echo "Could not find own node $REPLICA_ID among current voters (already removed?), nothing to do" + finished=true + fi + else + echo "Removing self would leave zero voters, skipping (this can't become safe later during my own termination -- nothing else will add a voter for me)" + finished=true + fi + [ "$finished" = true ] && break + else + echo "Could not identify any voters in the describe output (unrecognized format), skipping removal for safety and retrying..." + fi + fi + sleep 2 + done + if [ "$finished" != true ]; then + echo "ERROR: could not remove self (node $REPLICA_ID) from the voter set before terminating (every attempt within ${{DEADLINE}}s failed or the quorum was unreachable throughout); the on-disk voter set may now list this pod even though it is gone -- if nothing else corrects this, a later restart may get stuck and require manual recovery, see kraft-controller.adoc" + fi + exit 0 + "#, + deadline_seconds = pre_stop_deadline_seconds(graceful_shutdown_timeout), + binary = KAFKA_METADATA_QUORUM_BINARY, + config = ADMIN_CLIENT_PROPERTIES_PATH, + cli_timeout = CLI_CALL_TIMEOUT_SECONDS, + cli_kill_after = CLI_CALL_KILL_AFTER_SECONDS, + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, + extract_bootstrap_servers = extract_bootstrap_servers_command(), + ) } #[cfg(test)] mod tests { use super::*; + #[test] + fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() { + let command = quorum_manager_container_command(); + assert!(command.contains( + "grep '^controller.quorum.bootstrap.servers=' /stackable/config/controller.properties" + )); + assert!(command.contains(r#"--bootstrap-controller "$BOOTSTRAP_SERVERS""#)); + assert!(command.contains("add-controller")); + assert!(!command.contains("--bootstrap-controller 'localhost")); + assert!(!command.contains(r#"--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(); + assert!(command.contains("trap 'handle_term_signal' TERM")); + assert!(command.contains("sleep 10 &")); + assert!(command.contains("wait $!")); + } + + /// The whole point of backgrounding `add-controller`: a plain foreground `timeout ...` + /// call is not interrupted by an arriving `TERM` — bash only checks/runs traps between + /// commands or during the interruptible `wait` builtin — so a call already in flight when + /// the pod starts terminating could otherwise run to completion and race the `kafka` + /// container's own `preStop` removal, re-adding a pod that is simultaneously being + /// removed. Backgrounding it and having the trap actively `kill` it closes that window. + #[test] + fn quorum_manager_container_command_kills_an_in_flight_add_controller_attempt_on_term() { + let command = quorum_manager_container_command(); + assert!(command.contains("ADD_CONTROLLER_PID=$!")); + assert!(command.contains(r#"wait "$ADD_CONTROLLER_PID""#)); + assert!(command.contains(r#"kill -TERM "$ADD_CONTROLLER_PID""#)); + // The `add-controller` invocation itself must actually be backgrounded (not a plain + // foreground call) for the above to have any effect. + let add_controller_line = command + .lines() + .find(|line| line.contains("timeout") && line.contains("add-controller")) + .expect("the add-controller invocation is present"); + assert!( + add_controller_line.trim_end().ends_with('&'), + "add-controller must be backgrounded so TERM can interrupt `wait` immediately, \ + line was: {add_controller_line}" + ); + } + + /// 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(); + // 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 controller_remove_self_pre_stop_command_always_exits_zero() { + let command = controller_remove_self_pre_stop_command(None); + assert!(command.trim_end().ends_with("exit 0")); + assert!(command.contains("remove-controller")); + } + + /// A failed `remove-controller` attempt must not `break` out of the retry loop — unlike + /// the "already removed" and "would leave zero voters" cases, a failure is exactly the + /// situation the retry loop exists for. Only success (`finished=true` on that path) may + /// exit early. + #[test] + fn controller_remove_self_pre_stop_command_retries_a_failed_remove_controller_attempt() { + let command = controller_remove_self_pre_stop_command(None); + + // The loop's exit check is conditional on success (`finished=true`), not an + // unconditional `break` - so a failed attempt, which never reaches `finished=true`, + // falls through to the loop's retry instead of exiting immediately. + assert!(command.contains(r#"[ "$finished" = true ] && break"#)); + + // The failed-attempt branch itself must not set `finished=true` or `break` on its + // own - only the sibling success (`then`) branch does; this branch's only content is + // the log message. + let failure_message = "remove-controller attempt failed, will retry if time remains"; + let failure_message_pos = command + .find(failure_message) + .expect("the failure message is present in the generated script"); + let up_to_failure_message = &command[..failure_message_pos + failure_message.len()]; + let else_branch_start = up_to_failure_message + .rfind("else") + .expect("the failure message is inside an `else` branch"); + let else_branch = &up_to_failure_message[else_branch_start..]; + assert!(!else_branch.contains("finished=true")); + assert!(!else_branch.contains("break")); + } + + /// If every retry within `DEADLINE` fails, the script must say so loudly (an `ERROR:` + /// prefixed line, consistent with the `quorum-manager` main loop's own degraded-mode + /// messages) rather than silently letting the pod terminate with the voter never removed — + /// see `pre_stop_deadline_seconds`'s and this function's doc comments for why a silent + /// failure here is the specific gap that can strand a later restart-from-zero. + #[test] + fn controller_remove_self_pre_stop_command_logs_loudly_when_every_attempt_fails() { + let command = controller_remove_self_pre_stop_command(None); + assert!(command.contains(r#"if [ "$finished" != true ]; then"#)); + let error_branch = command + .split(r#"if [ "$finished" != true ]; then"#) + .nth(1) + .expect("the failure branch follows the retry loop"); + assert!(error_branch.contains("echo \"ERROR:")); + } + + /// The retry `DEADLINE` must be derived from the pod's actual `gracefulShutdownTimeout` + /// (via [`pre_stop_deadline_seconds`]) rather than hardcoded, and must stay within the + /// documented floor/cap regardless of how short or long that timeout is. + #[test] + fn pre_stop_deadline_seconds_is_derived_from_graceful_shutdown_timeout_within_floor_and_cap() { + // No configured timeout (shouldn't happen in practice) falls back to the floor. + assert_eq!( + pre_stop_deadline_seconds(None), + PRE_STOP_MIN_DEADLINE_SECONDS + ); + + // A short timeout (shorter than the reserved buffer) still gets at least the floor, + // never less than the original fixed behavior. + assert_eq!( + pre_stop_deadline_seconds(Some(Duration::from_secs(10))), + PRE_STOP_MIN_DEADLINE_SECONDS + ); + + // A generous timeout (the operator's own 30-minute default) is capped, not handed the + // entire budget minus the reserve. + assert_eq!( + pre_stop_deadline_seconds(Some(Duration::from_minutes_unchecked(30))), + PRE_STOP_MAX_DEADLINE_SECONDS + ); + + // A timeout comfortably between the floor and the cap (once the reserve is subtracted) + // is used as-is. + assert_eq!( + pre_stop_deadline_seconds(Some(Duration::from_secs(90))), + 90 - PRE_STOP_RESERVED_FOR_KAFKA_SHUTDOWN_SECONDS + ); + } + + /// The generated script's own `DEADLINE` must actually use + /// [`pre_stop_deadline_seconds`]'s output, not a literal left over from before it existed. + #[test] + fn controller_remove_self_pre_stop_command_deadline_reflects_the_configured_timeout() { + let command = + controller_remove_self_pre_stop_command(Some(Duration::from_minutes_unchecked(30))); + assert!(command.contains(&format!( + "DEADLINE=$((SECONDS + {}))", + PRE_STOP_MAX_DEADLINE_SECONDS + ))); + } + + /// 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 controller_remove_self_pre_stop_command_allows_removing_the_second_to_last_voter() { + let command = controller_remove_self_pre_stop_command(None); + 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" + ); + } + + /// Confirmed live: a controller pod that is the last remaining voter when it terminates + /// (e.g. scaling controllers down to 1, or the last survivor of a full teardown) hit the + /// "would leave zero voters" branch and, before this fix, kept retrying every 2s until + /// the full 25s `DEADLINE` elapsed for no benefit -- nothing else adds a voter for this + /// pod while it's terminating, so the outcome can never change. The branch must `break` + /// immediately instead of falling through to the loop's `sleep 2`. + #[test] + fn controller_remove_self_pre_stop_command_gives_up_immediately_on_the_last_voter() { + let command = controller_remove_self_pre_stop_command(None); + let after_zero_voters_message = command + .split("Removing self would leave zero voters") + .nth(1) + .expect("the zero-voters message is present in the generated script"); + let next_sleep = after_zero_voters_message + .find("sleep 2") + .expect("the loop's retry `sleep 2` follows somewhere after this branch"); + let until_next_retry = &after_zero_voters_message[..next_sleep]; + + assert!( + until_next_retry.contains("finished=true"), + "the zero-voters branch must mark the loop finished (no voter needs removing), \ + text was: {until_next_retry}" + ); + assert!( + until_next_retry.contains("break"), + "the zero-voters branch must break out of the retry loop immediately instead of \ + falling through to the loop's `sleep 2` and retrying until DEADLINE, text was: \ + {until_next_retry}" + ); + } + + /// 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(); + 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(); + // 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(); + assert!(command.contains("curl -s --max-time 5 --connect-timeout 2 localhost")); + } + + /// `timeout N cmd` (GNU coreutils, no `--kill-after`) only *sends* the signal after `N` + /// seconds — it does not force-kill the process, so if `cmd` doesn't honor the signal + /// promptly, the whole call can run far longer than `N` seconds. Confirmed directly, + /// independent of Kafka: `timeout 3 bash -c 'trap "" TERM; sleep 30'` takes the full 30s, + /// not 3s, while `timeout --kill-after=2 3 bash -c 'trap "" TERM; sleep 30'` is correctly + /// bounded to ~5s. This matters most for `controller_remove_self_pre_stop_command`, which runs + /// exactly when peers may be mid-termination (a blackholed, not actively-refused, + /// connection is exactly the kind of thing a JVM AdminClient can hang on past its own + /// `timeout` wrapper) — confirmed live (back when this ran as the `quorum-manager` + /// sidecar's own `preStop`, before it moved to the `kafka` container): during a full + /// namespace deletion, the `preStop` kept running for 100+ seconds, far past the script's + /// own ~25-40s design budget at the time. + #[test] + fn every_cli_call_has_a_kill_after_so_timeout_is_actually_enforced() { + let container_command = quorum_manager_container_command(); + let pre_stop_command = controller_remove_self_pre_stop_command(None); + + for command in [&container_command, &pre_stop_command] { + for line in command + .lines() + .filter(|line| line.contains(KAFKA_METADATA_QUORUM_BINARY)) + { + assert!( + line.contains("timeout --kill-after="), + "every kafka-metadata-quorum.sh invocation must use `timeout --kill-after=...` \ + so a hung call is actually bounded, not just signaled — offending line: {line}" + ); + } + } + } + + /// Builds a minimal [`KafkaPodDescriptor`] for the given role and replica. + /// + /// `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` — mirrors the identically-named helper in + /// `build/properties/mod.rs`'s own test module. + fn pod_descriptor(role: KafkaRole, replica: u16, node_id: u32) -> 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, + role, + client_port: 9093.into(), + } + } + + /// The controller with the lowest `node_id` bootstraps the quorum by itself + /// (`--standalone`); every other controller joins via the `quorum-manager` sidecar's + /// `add-controller` loop (`--no-initial-controllers`) — this is the runtime branch that + /// replaces baking a fixed `--initial-controllers ` into the format command. + #[test] + fn controller_kafka_container_command_branches_on_the_lowest_node_id() { + let descriptors = vec![ + pod_descriptor(KafkaRole::Controller, 0, 5), + pod_descriptor(KafkaRole::Controller, 1, 6), + pod_descriptor(KafkaRole::Controller, 2, 7), + ]; + let command = controller_kafka_container_command(descriptors); + + assert!(command.contains(r#"if [ "$REPLICA_ID" = "5" ]; then"#)); + assert!(command.contains("FORMAT_QUORUM_FLAG=--standalone")); + assert!(command.contains("FORMAT_QUORUM_FLAG=--no-initial-controllers")); + assert!(command.contains( + "bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/controller.properties --ignore-formatted \"$FORMAT_QUORUM_FLAG\"" + )); + // The old `--initial-controllers ` scheme is gone entirely, including its + // synthetic directory-id suffix. + assert!(!command.contains("--initial-controllers")); + assert!(!command.contains("0000000000-")); + } + + /// The whole point of removing the baked-in voter list: the container command must stay + /// byte-for-byte identical when only the *replica count* of an existing controller role + /// group changes (new replicas only ever get higher node ids), so scaling up/down no + /// longer forces Kubernetes to roll every already-existing controller pod just to pick up + /// an unchanged (`--ignore-formatted` no-ops it anyway) format command. + #[test] + fn controller_kafka_container_command_is_stable_across_replica_count_changes() { + let three_replicas = vec![ + pod_descriptor(KafkaRole::Controller, 0, 5), + pod_descriptor(KafkaRole::Controller, 1, 6), + pod_descriptor(KafkaRole::Controller, 2, 7), + ]; + let five_replicas = vec![ + pod_descriptor(KafkaRole::Controller, 0, 5), + pod_descriptor(KafkaRole::Controller, 1, 6), + pod_descriptor(KafkaRole::Controller, 2, 7), + pod_descriptor(KafkaRole::Controller, 3, 8), + pod_descriptor(KafkaRole::Controller, 4, 9), + ]; + + assert_eq!( + controller_kafka_container_command(three_replicas), + controller_kafka_container_command(five_replicas) + ); + } + + /// Brokers are never voters and never the bootstrap candidate — they always join (or, for + /// a fresh cluster, simply never assert any voter membership) via `--no-initial-controllers`. + #[test] + fn broker_start_command_always_uses_no_initial_controllers_in_kraft_mode() { + let command = broker_start_command(true); + assert!(command.contains("--no-initial-controllers")); + assert!(!command.contains("--initial-controllers")); + assert!(!command.contains("--standalone")); + } + #[test] fn test_constants() { // Test that dereferencing the constants does not panic. diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index 624525f71..bc54c3681 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -45,7 +45,6 @@ pub enum Error { pub fn add_kerberos_pod_config( kafka_security: &ValidatedKafkaSecurity, role: &KafkaRole, - cb_kcat_prober: &mut ContainerBuilder, cb_kafka: &mut ContainerBuilder, pb: &mut PodBuilder, ) -> Result<(), Error> { @@ -68,10 +67,9 @@ pub fn add_kerberos_pod_config( ) .context(AddVolumeSnafu)?; - for cb in [cb_kafka, cb_kcat_prober] { - cb.add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR) - .context(AddVolumeMountSnafu)?; - } + cb_kafka + .add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR) + .context(AddVolumeMountSnafu)?; } Ok(()) @@ -80,8 +78,8 @@ pub fn add_kerberos_pod_config( constant!(KRB5_CONFIG: EnvVarName = "KRB5_CONFIG"); constant!(KAFKA_OPTS: EnvVarName = "KAFKA_OPTS"); -/// The environment variables the Kerberos configuration requires on the Kafka and kcat-prober -/// containers, or an empty set when Kerberos is disabled. +/// The environment variables the Kerberos configuration requires on the Kafka container, or an +/// empty set when Kerberos is disabled. /// /// Returned as an [`EnvVarSet`] (rather than added to the containers directly) so the callers /// can merge the user's `envOverrides` on top, letting an override win on a name collision. diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index ad5d2b304..d3a6a0d80 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -229,13 +229,17 @@ pub(crate) fn role_group_selector( 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, STACKABLE_DATA_DIR, role::KafkaRole}, }; /// Sorted `metadata.name`s of the given resources, for order-independent assertions. @@ -278,6 +282,36 @@ mod tests { validated_cluster(&kafka) } + #[test] + fn build_succeeds_when_every_kraft_role_group_is_scaled_to_zero() { + 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: 0 + brokers: + roleGroups: + default: + replicas: 0 + "#, + ); + let cluster = validated_cluster(&kafka); + + build(&cluster).expect("build succeeds when the whole KRaft cluster is stopped"); + } + #[test] fn build_produces_expected_resource_names() { let cluster = kraft_mode_cluster(); @@ -365,6 +399,146 @@ mod tests { ); } + #[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:?}" + ); + // `add-controller` reads this controller's own on-disk `meta.properties` (written by + // `kafka-storage.sh format`, and pointed at by `log.dirs` in the merged config it + // connects with) to build the voter registration payload. Confirmed live: without + // this mount, every `add-controller` attempt fails with "Unable to read + // meta.properties from /stackable/data/kraft" — the path simply doesn't exist in + // this container without it. + assert!( + mount_paths.contains(&STACKABLE_DATA_DIR), + "the sidecar must mount the data directory holding its own meta.properties, or add-controller can never read its own identity, got: {mount_paths:?}" + ); + } + + /// Confirmed live: the `smoke-kraft` test cluster's admission control rejects any pod + /// whose memory limit-to-request ratio isn't exactly 1 ("memory max limit to request + /// ratio per Container is 1, but provided ratio is 2.000000"), which is also what the + /// operator's own `stackable_operator::builder::pod` warning already flags. Every + /// container's memory request must equal its memory limit, not just the `quorum-manager` + /// sidecar's — this test covers all containers in both the broker and controller pods so + /// a future container addition can't reintroduce this for either role. + #[test] + fn every_container_has_a_1_to_1_memory_limit_to_request_ratio() { + let cluster = kraft_mode_cluster(); + let resources = build(&cluster).expect("build succeeds"); + + for sts in &resources.stateful_sets { + let pod_spec = sts + .spec + .as_ref() + .expect("the StatefulSet should have a spec") + .template + .spec + .as_ref() + .expect("the pod template should have a spec"); + for container in &pod_spec.containers { + let resources = container + .resources + .as_ref() + .unwrap_or_else(|| panic!("container {} has no resources set", container.name)); + let request = resources + .requests + .as_ref() + .and_then(|r| r.get("memory")) + .unwrap_or_else(|| { + panic!("container {} has no memory request set", container.name) + }); + let limit = resources + .limits + .as_ref() + .and_then(|l| l.get("memory")) + .unwrap_or_else(|| { + panic!("container {} has no memory limit set", container.name) + }); + assert_eq!( + request, limit, + "container {} must have memory request == memory limit (ratio 1:1); \ + the smoke-kraft test cluster's admission control rejects anything else", + container.name + ); + } + } + } + + /// 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 2f3951f57..3ce39b94e 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")] @@ -56,18 +61,28 @@ pub fn uses_legacy_log4j(product_version: &str) -> bool { product_version.starts_with("3.") } +/// `controller.quorum.bootstrap.servers` addresses, one per distinct controller role group, +/// pointing at each role group's own headless Service DNS name rather than individual pod +/// FQDNs. +/// +/// Only adding or removing a whole role group changes this list. pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Vec { - pod_descriptors + let mut role_group_addresses: Vec = pod_descriptors .iter() .filter(|pd| pd.role == KafkaRole::Controller) .map(|desc| { format!( - "{fqdn}:{client_port}", - fqdn = desc.fqdn(), - client_port = desc.client_port + "{service}.{namespace}.svc.{cluster_domain}:{client_port}", + service = desc.role_group_service_name, + namespace = desc.namespace, + cluster_domain = desc.cluster_domain, + client_port = desc.client_port, ) }) - .collect::>() + .collect(); + role_group_addresses.sort(); + role_group_addresses.dedup(); + role_group_addresses } #[cfg(test)] @@ -86,8 +101,128 @@ 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(), + } + } + + /// `kraft_controllers` points at the controller role group's *headless Service* DNS name + /// (no pod prefix), not individual pod FQDNs. A headless Service's own DNS name resolves + /// to every backing pod's IP (Kafka's own AdminClient default, + /// `client.dns.lookup=use_all_dns_ips`, already expects exactly this), and the + /// operator's headless Service sets `publishNotReadyAddresses: true`, so this also works + /// during initial cluster formation before any pod is Ready. This is what makes + /// `controller.quorum.bootstrap.servers` invariant to the controller role group's + /// replica count: adding or removing replicas within an existing role group never + /// changes the role group's own Service name. + #[test] + fn kraft_controllers_points_at_the_role_group_headless_service_not_individual_pods() { + let pod_descriptors = vec![ + pod_descriptor(KafkaRole::Controller, 0, 9093), + pod_descriptor(KafkaRole::Controller, 1, 9093), + pod_descriptor(KafkaRole::Controller, 2, 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-headless.default.svc.cluster.local:9093" + ); + } + + /// The whole point: scaling an existing controller role group up or down must not change + /// `kraft_controllers`'s output at all, since it no longer depends on which replicas + /// currently exist — the role group's Service name is stable regardless. + #[test] + fn kraft_controllers_is_stable_across_replica_count_changes() { + let three_replicas = vec![ + pod_descriptor(KafkaRole::Controller, 0, 9093), + pod_descriptor(KafkaRole::Controller, 1, 9093), + pod_descriptor(KafkaRole::Controller, 2, 9093), + ]; + let five_replicas = vec![ + pod_descriptor(KafkaRole::Controller, 0, 9093), + pod_descriptor(KafkaRole::Controller, 1, 9093), + pod_descriptor(KafkaRole::Controller, 2, 9093), + pod_descriptor(KafkaRole::Controller, 3, 9093), + pod_descriptor(KafkaRole::Controller, 4, 9093), + ]; + + assert_eq!( + kraft_controllers(&three_replicas), + kraft_controllers(&five_replicas) + ); + } + + /// Multiple controller role groups each have their own headless Service, so each must + /// still get its own bootstrap-servers entry — deduplication is per-Service, not a + /// blanket "collapse everything to one entry". + #[test] + fn kraft_controllers_lists_every_distinct_role_groups_service_once() { + let mut default_group_pod = pod_descriptor(KafkaRole::Controller, 0, 9093); + let mut other_group_pod = pod_descriptor(KafkaRole::Controller, 0, 9093); + other_group_pod.role_group_statefulset_name = "kafka-controller-other" + .parse() + .expect("valid statefulset name"); + other_group_pod.role_group_service_name = "kafka-controller-other-headless" + .parse() + .expect("valid service name"); + // Second replica of the *same* role group as `default_group_pod` — must not produce + // a second entry for that Service. + let default_group_pod_replica_1 = { + let mut pod = pod_descriptor(KafkaRole::Controller, 1, 9093); + pod.node_id = 1; + pod + }; + default_group_pod.node_id = 0; + + let pod_descriptors = vec![ + default_group_pod, + default_group_pod_replica_1, + other_group_pod, + ]; + + let mut quorum_bootstrap_servers = kraft_controllers(&pod_descriptors); + quorum_bootstrap_servers.sort(); + + assert_eq!( + quorum_bootstrap_servers, + vec![ + "kafka-controller-default-headless.default.svc.cluster.local:9093".to_string(), + "kafka-controller-other-headless.default.svc.cluster.local:9093".to_string(), + ] + ); + } } 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 a1be2a907..49c4e2616 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -18,7 +18,7 @@ use crate::{ ConfigFileName, config_file_name, product_logging::role_group_config_map_data, }, recommended_labels_for_role_group_resources, - security::client_properties, + security::{client_properties, controller_admin_client_properties}, }, }, crd::{ @@ -51,13 +51,20 @@ 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, }, - - #[snafu(display("no Kraft controllers found to build"))] - NoKraftControllersFound, } /// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator. @@ -87,10 +94,6 @@ pub fn build_rolegroup_config_map( .pod_descriptors(None) .context(BuildPodDescriptorsSnafu)?; - if cluster_config.is_kraft_mode() && pod_descriptors.is_empty() { - return NoKraftControllersFoundSnafu.fail(); - } - let kafka_config = match &validated_rg.config.config { AnyConfig::Broker(_) => crate::controller::build::properties::broker_properties::build( cluster_config, @@ -164,7 +167,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(), })?, ) @@ -177,6 +180,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/mod.rs b/rust/operator-binary/src/controller/build/resource/mod.rs index 2d0977ad7..7f458714f 100644 --- a/rust/operator-binary/src/controller/build/resource/mod.rs +++ b/rust/operator-binary/src/controller/build/resource/mod.rs @@ -4,6 +4,7 @@ pub mod config_map; pub mod discovery; pub mod listener; pub mod pdb; +pub mod probes; pub mod rbac; pub mod service; pub mod statefulset; diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs new file mode 100644 index 000000000..8b8a5ad69 --- /dev/null +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -0,0 +1,165 @@ +//! Container probes for the Kafka `kafka` container (broker and controller roles). + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + builder::pod::probe::{self, ProbeBuilder}, + k8s_openapi::{ + api::core::v1::{Probe, TCPSocketAction}, + apimachinery::pkg::util::intstr::IntOrString, + }, + shared::time::Duration, + v2::types::common::Port, +}; + +use crate::controller::{ + build::security::kcat_prober_container_commands, security::ValidatedKafkaSecurity, +}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build the {name} probe"))] + BuildProbe { + source: probe::Error, + name: &'static str, + }, +} + +/// The broker `kafka` container's readiness probe. +/// +/// Uses `kcat` rather than the official Kafka tools, since they incur a lot of unacceptable perf +/// overhead when run repeatedly as a probe. Only allow the global load balancing service to send +/// traffic to pods that are members of the quorum. This also acts as a hint to the StatefulSet +/// controller to wait for each pod to enter quorum before taking down the next. +pub fn broker_kcat_readiness_probe( + kafka_security: &ValidatedKafkaSecurity, +) -> Result { + ProbeBuilder::exec_command( + // If the broker is able to get its fellow cluster members then it has at least + // completed basic registration at some point + kcat_prober_container_commands(kafka_security), + ) + .with_period(Duration::from_secs(2)) + .with_timeout(Duration::from_secs(5)) + // `ProbeBuilder` otherwise defaults this to 1; kept at Kubernetes' own default (3) to match + // the pre-`ProbeBuilder` behaviour, which left this field unset. + .with_failure_threshold(3) + .build() + .context(BuildProbeSnafu { + name: "kcat readiness", + }) +} + +/// A `Probe` combining a plain TCP check of the broker's client listener with a check that the +/// broker's own JMX `BrokerState` metric reports `RUNNING` (state `3`). +/// +/// Used for both the `startupProbe` (so the `livenessProbe` doesn't start counting failures +/// until the broker has actually finished starting - the client port can accept connections +/// before the broker reaches `RUNNING`, e.g. while still replaying its log) and the +/// `livenessProbe` (restarting a broker stuck in some other state, e.g. `RECOVERY` after a +/// crash); only the timing parameters differ between the two uses. +pub fn broker_running_probe( + client_port: Port, + metrics_port: Port, + timeout_seconds: u64, + period_seconds: u64, + failure_threshold: i32, +) -> Result { + ProbeBuilder::exec_command([ + "bash".to_string(), + "-c".to_string(), + format!( + "timeout 2 bash -c 'cat < /dev/null > /dev/tcp/localhost/{client_port}' || exit 1\n\ + curl -s --max-time 2 localhost:{metrics_port}/metrics | grep -qE 'kafka_server_kafkaserver_brokerstate 3(\\.0)?$'" + ), + ]) + .with_period(Duration::from_secs(period_seconds)) + .with_timeout(Duration::from_secs(timeout_seconds)) + .with_failure_threshold(failure_threshold) + .build() + .context(BuildProbeSnafu { + name: "broker running", + }) +} + +/// A `Probe` that dials the controller's KRaft listener socket via a plain TCP connect. +/// +/// This only proves the socket is open, not that the node has a healthy Raft state (leader, +/// follower, or voted). Used for `startupProbe`: there is no meaningful Raft state to check +/// yet while the process is still starting, so a bare TCP check is all that's meaningful this +/// early. +pub fn controller_tcp_probe( + port: Port, + timeout_seconds: u64, + period_seconds: u64, + failure_threshold: i32, +) -> Result { + ProbeBuilder::tcp_socket(TCPSocketAction { + port: IntOrString::Int(port.into()), + ..Default::default() + }) + .with_period(Duration::from_secs(period_seconds)) + .with_timeout(Duration::from_secs(timeout_seconds)) + .with_failure_threshold(failure_threshold) + .build() + .context(BuildProbeSnafu { + name: "controller startup", + }) +} + +/// A `Probe` that curls the JMX Prometheus exporter's `/metrics` endpoint and checks that the +/// controller's Raft state is one of the healthy states (`leader`, `follower`, or `voted`) +/// rather than stuck in `unattached` or `candidate`. +pub fn controller_raft_state_probe( + metrics_port: Port, + timeout_seconds: u64, + period_seconds: u64, + failure_threshold: i32, +) -> Result { + ProbeBuilder::exec_command([ + "bash".to_string(), + "-c".to_string(), + format!( + "curl -s localhost:{metrics_port}/metrics | grep -E 'kafka_server_raft_metrics_current_state\\{{state=\"(leader|follower|voted)\",?\\}}'" + ), + ]) + .with_period(Duration::from_secs(period_seconds)) + .with_timeout(Duration::from_secs(timeout_seconds)) + .with_failure_threshold(failure_threshold) + .build() + .context(BuildProbeSnafu { + name: "controller raft state", + }) +} + +/// A `Probe` combining a plain TCP check of the controller's KRaft listener with a check that +/// its local Raft state isn't stuck in `unattached`. +/// +/// This is needed to work around a bug in KRaft where a new controller is stuck in a loop +/// trying to fetch Raft metadata from its self. +/// +/// This can happen when the headless service used to point to the bootstrap controllers +/// happens to resolve to this exact pod. +pub fn controller_stuck_unattached_liveness_probe( + client_port: Port, + metrics_port: Port, + timeout_seconds: u64, + period_seconds: u64, + failure_threshold: i32, +) -> Result { + ProbeBuilder::exec_command([ + "bash".to_string(), + "-c".to_string(), + format!( + "timeout 2 bash -c 'cat < /dev/null > /dev/tcp/localhost/{client_port}' || exit 1\n\ + state=$(curl -s --max-time 2 localhost:{metrics_port}/metrics | grep -oE 'kafka_server_raft_metrics_current_state\\{{state=\"[a-z]+\"\\}}' | grep -oE '\"[a-z]+\"' | tr -d '\"')\n\ + [ \"$state\" != \"unattached\" ]" + ), + ]) + .with_period(Duration::from_secs(period_seconds)) + .with_timeout(Duration::from_secs(timeout_seconds)) + .with_failure_threshold(failure_threshold) + .build() + .context(BuildProbeSnafu { + name: "controller stuck-unattached liveness", + }) +} diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 063d62325..00adff25c 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -19,11 +19,10 @@ use stackable_operator::{ api::{ apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetUpdateStrategy}, core::v1::{ - ConfigMapVolumeSource, ContainerPort, EnvVar, ExecAction, PodSpec, Probe, - TCPSocketAction, Volume, + ConfigMapVolumeSource, ContainerPort, EnvVar, ExecAction, LifecycleHandler, Volume, }, }, - apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, + apimachinery::pkg::apis::meta::v1::LabelSelector, }, product_logging, v2::{ @@ -43,13 +42,15 @@ use stackable_operator::{ }, }; +use super::probes; use crate::{ controller::{ RoleGroupName, ValidatedCluster, ValidatedRoleGroupConfig, build::{ command::{ KAFKA_LOG4J_OPTS, broker_kafka_container_commands, - controller_kafka_container_command, kafka_log_opts, + controller_kafka_container_command, controller_remove_self_pre_stop_command, + kafka_log_opts, quorum_manager_container_command, }, graceful_shutdown::add_graceful_shutdown_config, kerberos::{add_kerberos_pod_config, kerberos_env_vars}, @@ -57,8 +58,8 @@ use crate::{ recommended_labels_for_role_group_resources, recommended_labels_for_unversioned_role_group_resources, role_group_selector, 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, }, }, node_id_hasher::node_id_hash32_offset, @@ -89,7 +90,6 @@ stackable_operator::constant!(KAFKA_CLIENT_PORT: EnvVarName = "KAFKA_CLIENT_PORT stackable_operator::constant!(NAMESPACE: EnvVarName = "NAMESPACE"); stackable_operator::constant!(ROLEGROUP_HEADLESS_SERVICE_NAME: EnvVarName = "ROLEGROUP_HEADLESS_SERVICE_NAME"); stackable_operator::constant!(CLUSTER_DOMAIN: EnvVarName = "CLUSTER_DOMAIN"); -stackable_operator::constant!(PRE_STOP_CONTROLLER_SLEEP_SECONDS: EnvVarName = "PRE_STOP_CONTROLLER_SLEEP_SECONDS"); stackable_operator::constant!(EXTRA_ARGS: EnvVarName = "EXTRA_ARGS"); // Needed for the `containerdebug` process to log its tracing information to. stackable_operator::constant!(CONTAINERDEBUG_LOG_DIRECTORY: EnvVarName = "CONTAINERDEBUG_LOG_DIRECTORY"); @@ -128,7 +128,32 @@ 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 caller merges the user's `envOverrides` on top (so a user override wins on a name +/// collision); the `quorum-manager` sidecar additionally gets its own container-specific env +/// vars layered on top (see [`KAFKA_NODE_ID_OFFSET`]). +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(&NAMESPACE, &FieldPathEnvVar::Namespace) + .with_value( + &ROLEGROUP_HEADLESS_SERVICE_NAME, + resource_names.headless_service_name().to_string(), + ) + .with_value( + &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 { @@ -162,6 +187,9 @@ pub enum Error { source: crate::controller::PodDescriptorsError, }, + #[snafu(display("failed to build container probe"))] + BuildProbe { source: probes::Error }, + #[snafu(display("failed to construct JVM arguments"))] ConstructJvmArguments { source: crate::controller::build::jvm::Error, @@ -206,12 +234,6 @@ pub fn build_broker_rolegroup_statefulset( role_group_name, ); - let kcat_prober_container_name = BrokerContainer::KcatProber.to_string(); - let mut cb_kcat_prober = - ContainerBuilder::new(&kcat_prober_container_name).context(InvalidContainerNameSnafu { - name: kcat_prober_container_name.clone(), - })?; - let kafka_container_name = BrokerContainer::Kafka.to_string(); let mut cb_kafka = ContainerBuilder::new(&kafka_container_name).context(InvalidContainerNameSnafu { @@ -228,7 +250,6 @@ pub fn build_broker_rolegroup_statefulset( add_broker_volume_and_volume_mounts( kafka_security, &mut pod_builder, - &mut cb_kcat_prober, &mut cb_kafka, &requested_secret_lifetime, ) @@ -249,14 +270,8 @@ pub fn build_broker_rolegroup_statefulset( )); if kafka_security.has_kerberos_enabled() { - add_kerberos_pod_config( - kafka_security, - kafka_role, - &mut cb_kcat_prober, - &mut cb_kafka, - &mut pod_builder, - ) - .context(AddKerberosConfigSnafu)?; + add_kerberos_pod_config(kafka_security, kafka_role, &mut cb_kafka, &mut pod_builder) + .context(AddKerberosConfigSnafu)?; } // Operator-set env vars first; the user's `envOverrides` are merged on top last and win. @@ -274,6 +289,18 @@ pub fn build_broker_rolegroup_statefulset( .merge(validated_rg.env_overrides.clone()) .into(); + // The client port can accept connections before the broker has replayed its log and + // reached the JMX `RUNNING` state, so the startupProbe waits for both, giving it up to + // 5 minutes (60 * 5s) before the livenessProbe is allowed to start counting failures. + let broker_startup_probe = + probes::broker_running_probe(kafka_security.client_port(), METRICS_PORT, 5, 5, 60) + .context(BuildProbeSnafu)?; + let broker_liveness_probe = + probes::broker_running_probe(kafka_security.client_port(), METRICS_PORT, 10, 30, 20) + .context(BuildProbeSnafu)?; + let broker_readiness_probe = + probes::broker_kcat_readiness_probe(kafka_security).context(BuildProbeSnafu)?; + cb_kafka .image_from_product_image(resolved_product_image) .command(vec![ @@ -285,12 +312,7 @@ pub fn build_broker_rolegroup_statefulset( ]) .args(vec![broker_kafka_container_commands( validated_cluster.cluster_config.is_kraft_mode(), - // we need controller pods - validated_cluster - .pod_descriptors(Some(&KafkaRole::Controller)) - .context(BuildPodDescriptorsSnafu)?, kafka_security, - &resolved_product_image.product_version, )]); cb_kafka @@ -311,44 +333,10 @@ pub fn build_broker_rolegroup_statefulset( .context(AddVolumeMountSnafu)? .add_volume_mount(STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) .context(AddVolumeMountSnafu)? - .resources(merged_config.resources().clone().into()); - - // Use kcat sidecar for probing container status rather than the official Kafka tools, since they incur a lot of - // unacceptable perf overhead - cb_kcat_prober - .image_from_product_image(resolved_product_image) - .command(vec!["sleep".to_string(), "infinity".to_string()]) - .add_env_vars(Vec::::from( - EnvVarSet::new() - .with_field_path(&POD_NAME, &FieldPathEnvVar::Name) - .merge(kerberos_env_vars(kafka_security)), - )) - .resources( - ResourceRequirementsBuilder::new() - .with_cpu_request("100m") - .with_cpu_limit("200m") - .with_memory_request("128Mi") - .with_memory_limit("128Mi") - .build(), - ) - .add_volume_mount( - LISTENER_BOOTSTRAP_VOLUME_NAME, - STACKABLE_LISTENER_BOOTSTRAP_DIR, - ) - .context(AddVolumeMountSnafu)? - .add_volume_mount(LISTENER_BROKER_VOLUME_NAME, STACKABLE_LISTENER_BROKER_DIR) - .context(AddVolumeMountSnafu)? - // Only allow the global load balancing service to send traffic to pods that are members of the quorum - // This also acts as a hint to the StatefulSet controller to wait for each pod to enter quorum before taking down the next - .readiness_probe(Probe { - exec: Some(ExecAction { - // If the broker is able to get its fellow cluster members then it has at least completed basic registration at some point - command: Some(kcat_prober_container_commands(kafka_security)), - }), - timeout_seconds: Some(5), - period_seconds: Some(2), - ..Probe::default() - }); + .resources(merged_config.resources().clone().into()) + .startup_probe(broker_startup_probe) + .liveness_probe(broker_liveness_probe) + .readiness_probe(broker_readiness_probe); add_log_config_volume( &mut pod_builder, @@ -390,7 +378,6 @@ pub fn build_broker_rolegroup_statefulset( .metadata(metadata) .image_pull_secrets_from_product_image(resolved_product_image) .add_container(cb_kafka.build()) - .add_container(cb_kcat_prober.build()) .affinity(&merged_config.affinity); add_common_pod_config( @@ -414,10 +401,6 @@ pub fn build_broker_rolegroup_statefulset( let mut pod_template = pod_builder.build_template(); - let pod_template_spec = pod_template.spec.get_or_insert_with(PodSpec::default); - // Don't run kcat pod as PID 1, to ensure that default signal handlers apply - pod_template_spec.share_process_namespace = Some(true); - // Pod overrides were already merged (role <- role group) during validation. pod_template.merge_from(validated_rg.pod_overrides.clone()); @@ -473,19 +456,17 @@ 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 last and win. - let env: Vec = common_operator_env_vars(validated_cluster, kafka_security) - .with_field_path(&NAMESPACE, &FieldPathEnvVar::Namespace) - .with_value( - &ROLEGROUP_HEADLESS_SERVICE_NAME, - resource_names.headless_service_name().to_string(), - ) - .with_value( - &CLUSTER_DOMAIN, - validated_cluster.cluster_domain.to_string(), - ) - .with_value(&PRE_STOP_CONTROLLER_SLEEP_SECONDS, "10") + // 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() .merge(common_kafka_env( merged_config, &validated_rg @@ -498,6 +479,36 @@ pub fn build_controller_rolegroup_statefulset( .merge(validated_rg.env_overrides.clone()) .into(); + let quorum_manager_env: Vec = controller_shared_env + .with_value(&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)?; + + // The controller listener socket only opens once the KRaft node has finished replaying + // its metadata log, which can take a while on a slow first boot or after a long outage. + // The startupProbe gives it up to 5 minutes (60 * 5s) before the liveness probe is + // allowed to start counting failures at all, so a slow (but progressing) boot is never + // mistaken for a stuck process. + let controller_startup_probe = + probes::controller_tcp_probe(kafka_security.client_port(), 5, 5, 60) + .context(BuildProbeSnafu)?; + // See `probes::controller_stuck_unattached_liveness_probe`'s doc comment for why this is no + // longer a plain TCP check. + let controller_liveness_probe = probes::controller_stuck_unattached_liveness_probe( + kafka_security.client_port(), + METRICS_PORT, + 10, + 30, + 20, + ) + .context(BuildProbeSnafu)?; + let controller_readiness_probe = + probes::controller_raft_state_probe(METRICS_PORT, 10, 10, 6).context(BuildProbeSnafu)?; + cb_kafka .image_from_product_image(resolved_product_image) .command(vec![ @@ -508,10 +519,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, )]); cb_kafka @@ -526,27 +534,26 @@ pub fn build_controller_rolegroup_statefulset( .add_volume_mount(STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) .context(AddVolumeMountSnafu)? .resources(merged_config.resources().clone().into()) - // TODO: improve probes - .liveness_probe(Probe { - tcp_socket: Some(TCPSocketAction { - port: IntOrString::Int(kafka_security.client_port().into()), - ..Default::default() - }), - timeout_seconds: Some(10), - period_seconds: Some(10), - failure_threshold: Some(6), - ..Probe::default() - }) - .readiness_probe(Probe { - tcp_socket: Some(TCPSocketAction { - port: IntOrString::Int(kafka_security.client_port().into()), - ..Default::default() + .startup_probe(controller_startup_probe) + .liveness_probe(controller_liveness_probe) + .readiness_probe(controller_readiness_probe); + // Skipped when Kerberos is enabled, matching `build_quorum_manager_container`'s own + // gating — `admin-client.properties` (the file this removal call relies on) only covers + // the TLS/SSL case. + if !kafka_security.has_kerberos_enabled() { + cb_kafka.lifecycle_pre_stop(LifecycleHandler { + exec: Some(ExecAction { + command: Some(vec![ + "/bin/bash".to_string(), + "-c".to_string(), + controller_remove_self_pre_stop_command( + merged_config.graceful_shutdown_timeout, + ), + ]), }), - timeout_seconds: Some(10), - period_seconds: Some(10), - failure_threshold: Some(6), - ..Probe::default() + ..LifecycleHandler::default() }); + } add_log_config_volume( &mut pod_builder, @@ -579,6 +586,12 @@ 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_manager_env)? + { + pod_builder.add_container(quorum_manager_container); + } + add_common_pod_config( &mut pod_builder, &resource_names, @@ -616,7 +629,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() @@ -664,8 +677,6 @@ fn container_ports(kafka_security: &ValidatedKafkaSecurity) -> Vec 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, + 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(), + ]) + .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.to_string(), "-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 budgets + .with_cpu_limit("500m") + .with_memory_request("512Mi") + .with_memory_limit("512Mi") + .build(), + ) + .add_volume_mount(STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) + .context(AddVolumeMountSnafu)? + // `controller_admin_client_properties` 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)? + // `add-controller` reads this controller's own on-disk `meta.properties` (its + // `node.id`/`directory.id`, written by `kafka-storage.sh format`) from `log.dirs` in + // the merged config it connects with - without this mount, every + // `add-controller` attempt failed with "Unable to read meta.properties from + // /stackable/data/kraft", since that path doesn't exist in this container's + // filesystem at all without it. This mounts the *same* per-pod PVC the `kafka` + // container itself writes `meta.properties` into, read-write for parity with it + // (the CLI tool doesn't document a read-only requirement, and this repo has no + // read-only-mount helper to reach for). + .add_volume_mount(LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) + .context(AddVolumeMountSnafu)?; + + Ok(Some(cb.build())) +} + fn add_vector_container( pod_builder: &mut PodBuilder, vector_container_name: &ContainerName, @@ -803,6 +886,8 @@ fn add_vector_container( #[cfg(test)] mod tests { + use stackable_operator::k8s_openapi::apimachinery::pkg::util::intstr::IntOrString; + use super::*; use crate::controller::test_support::{minimal_kafka, validated_cluster}; @@ -816,7 +901,6 @@ mod tests { let _ = *NAMESPACE; let _ = *ROLEGROUP_HEADLESS_SERVICE_NAME; let _ = *CLUSTER_DOMAIN; - let _ = *PRE_STOP_CONTROLLER_SLEEP_SECONDS; let _ = *EXTRA_ARGS; let _ = *CONTAINERDEBUG_LOG_DIRECTORY; let _ = *ZOOKEEPER; @@ -888,10 +972,11 @@ mod tests { assert_eq!(containerdebug[0].value.as_deref(), Some("/custom/log/dir")); } - /// Same guarantee for the controller role, whose env vars are assembled by a separate - /// builder ([`build_controller_rolegroup_statefulset`]). - #[test] - fn controller_env_overrides_override_operator_set_env_vars() { + /// A minimal KRaft cluster with one controller role group, resolved through the real + /// validate step (mirroring the fixtures in `build/mod.rs`'s own tests), since + /// `ValidatedCluster` carries several resolved types that are impractical to construct by + /// hand. + fn kraft_mode_cluster() -> crate::controller::ValidatedCluster { let kafka = minimal_kafka( r#" apiVersion: kafka.stackable.tech/v1alpha1 @@ -915,13 +1000,20 @@ mod tests { replicas: 3 "#, ); - let cluster = validated_cluster(&kafka); + validated_cluster(&kafka) + } + + /// Same guarantee for the controller role, whose env vars are assembled by a separate + /// builder ([`build_controller_rolegroup_statefulset`]). + #[test] + fn controller_env_overrides_override_operator_set_env_vars() { + let cluster = kraft_mode_cluster(); let role_group_name = RoleGroupName::from_str("default").expect("valid role group name"); let mut validated_rg = cluster.role_group_configs[&KafkaRole::Controller][&role_group_name].clone(); validated_rg.env_overrides = validated_rg .env_overrides - .with_value(&PRE_STOP_CONTROLLER_SLEEP_SECONDS, "42"); + .with_value(&CONTAINERDEBUG_LOG_DIRECTORY, "/custom/log/dir"); let stateful_set = build_controller_rolegroup_statefulset( &KafkaRole::Controller, @@ -944,15 +1036,549 @@ mod tests { .env .expect("the kafka container has env vars"); - let sleep_seconds: Vec<_> = env + let containerdebug: Vec<_> = env .iter() - .filter(|env_var| env_var.name == "PRE_STOP_CONTROLLER_SLEEP_SECONDS") + .filter(|env_var| env_var.name == "CONTAINERDEBUG_LOG_DIRECTORY") .collect(); assert_eq!( - sleep_seconds.len(), + containerdebug.len(), 1, "the override must replace the operator-set value, not duplicate it" ); - assert_eq!(sleep_seconds[0].value.as_deref(), Some("42")); + assert_eq!(containerdebug[0].value.as_deref(), Some("/custom/log/dir")); + } + + #[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()) + ); + } + + /// End-to-end regression covering the whole point of removing `--initial-controllers` + /// (and the sidecar's own baked-in bootstrap-servers literal) from the controller pod + /// template: scaling an existing controller role group's replica count must not change + /// either container's `command`, or Kubernetes will roll every already-existing + /// controller pod on every scale-up/down, not just the ones actually being added or + /// removed. Confirmed live: before this fix, both the `kafka` container's format command + /// and the `quorum-manager` sidecar's bootstrap-servers literal changed with replica + /// count, forcing a full rolling restart on every scale operation. + #[test] + fn controller_pod_template_is_stable_across_replica_count_changes() { + let three_replicas = kraft_mode_cluster(); + let five_replicas = crate::controller::test_support::validated_cluster( + &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.9.2 + clusterConfig: + metadataManager: kraft + controllers: + roleGroups: + default: + replicas: 5 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ), + ); + + let three_containers = controller_containers(&three_replicas); + let five_containers = controller_containers(&five_replicas); + + for name in ["kafka", QUORUM_MANAGER_CONTAINER_NAME] { + let three_command = three_containers + .iter() + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("the {name} container is built (3 replicas)")) + .command + .clone(); + let five_command = five_containers + .iter() + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("the {name} container is built (5 replicas)")) + .command + .clone(); + assert_eq!( + three_command, five_command, + "the {name} container's command must not change when only the replica count \ + of an existing controller role group changes" + ); + } + } + + 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")); + + // The sidecar only ever joins the quorum now — it has no `preStop` hook of its own. + // See `controller_kafka_container_has_a_remove_self_pre_stop_hook` for why the + // removal-on-departure half moved to the `kafka` container instead. + assert!(sidecar.lifecycle.is_none()); + } + + /// `remove-controller` must run as the `kafka` container's own `preStop` hook, not the + /// `quorum-manager` sidecar's — `preStop` only delays *that same container's* `SIGTERM`, + /// and it's the `kafka` container's own Raft process (the thing actually leaving the + /// voter set) that needs to stay alive while removal is attempted. See + /// `controller_remove_self_pre_stop_command`'s doc comment for the full rationale. + #[test] + fn controller_kafka_container_has_a_remove_self_pre_stop_hook() { + let cluster = kraft_mode_cluster(); + let container = controller_kafka_container(&cluster); + + let pre_stop_command = container + .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 kafka container 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 `add-controller` main loop silently (an unset + // variable under `set -u` aborts the script). + let node_id_offset_name = KAFKA_NODE_ID_OFFSET.to_string(); + assert!( + sidecar_env_names.contains(&node_id_offset_name.as_str()), + "quorum-manager sidecar is missing the {node_id_offset_name} 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, 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) + ); + } + + fn controller_kafka_container( + cluster: &crate::controller::ValidatedCluster, + ) -> stackable_operator::k8s_openapi::api::core::v1::Container { + controller_containers(cluster) + .into_iter() + .find(|c| c.name == "kafka") + .expect("the kafka container is built") + } + + fn broker_kafka_container( + cluster: &crate::controller::ValidatedCluster, + ) -> stackable_operator::k8s_openapi::api::core::v1::Container { + 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"); + sts.spec + .expect("the StatefulSet has a spec") + .template + .spec + .expect("the pod template has a spec") + .containers + .into_iter() + .find(|c| c.name == "kafka") + .expect("the kafka container is built") + } + + #[test] + fn broker_kafka_container_has_a_startup_probe() { + let cluster = kraft_mode_cluster(); + let container = broker_kafka_container(&cluster); + let client_port = cluster.cluster_config.kafka_security.client_port(); + + let startup_probe = container + .startup_probe + .expect("the broker kafka container must have a startupProbe"); + let exec = startup_probe + .exec + .expect("the startupProbe must be an exec check, not a bare tcpSocket check"); + let command = exec.command.expect("exec has a command"); + let script = command.last().expect("the exec command has a script arg"); + + assert!( + script.contains(&format!("/dev/tcp/localhost/{client_port}")), + "expected a TCP reachability check against the broker's own client port, script was: {script}" + ); + assert!( + script.contains("kafka_server_kafkaserver_brokerstate 3"), + "expected a check for the broker's JMX BrokerState metric being RUNNING (3), \ + script was: {script}" + ); + assert_eq!(startup_probe.timeout_seconds, Some(5)); + assert_eq!(startup_probe.period_seconds, Some(5)); + assert_eq!(startup_probe.failure_threshold, Some(60)); + } + + /// The liveness probe must check both TCP reachability (a genuinely dead/hung process must + /// still be restarted) and that the broker's JMX `BrokerState` metric reports `RUNNING` + /// (state `3`) - see `probes::broker_running_probe`'s doc comment for why the same check + /// backs both the startup and liveness probes. + #[test] + fn broker_kafka_container_liveness_probe_checks_tcp_and_running_state() { + let cluster = kraft_mode_cluster(); + let container = broker_kafka_container(&cluster); + let client_port = cluster.cluster_config.kafka_security.client_port(); + + let liveness_probe = container + .liveness_probe + .expect("the broker kafka container must have a livenessProbe"); + let exec = liveness_probe + .exec + .expect("the livenessProbe must be an exec check, not a bare tcpSocket check"); + let command = exec.command.expect("exec has a command"); + let script = command.last().expect("the exec command has a script arg"); + + assert!( + script.contains(&format!("/dev/tcp/localhost/{client_port}")), + "expected a TCP reachability check against the broker's own client port, script was: {script}" + ); + assert!( + script.contains("kafka_server_kafkaserver_brokerstate 3"), + "expected a check for the broker's JMX BrokerState metric being RUNNING (3), \ + script was: {script}" + ); + + assert_eq!(liveness_probe.timeout_seconds, Some(10)); + assert_eq!(liveness_probe.period_seconds, Some(30)); + assert_eq!(liveness_probe.failure_threshold, Some(20)); + } + + /// The `kcat`-based readiness probe runs directly on the `kafka` container - there is no + /// separate `kcat-prober` sidecar (removed since `kcat` ships in the same product image the + /// `kafka` container already uses, so a dedicated container was no longer needed). + #[test] + fn broker_kafka_container_readiness_probe_uses_kcat() { + let cluster = kraft_mode_cluster(); + let container = broker_kafka_container(&cluster); + + let readiness_probe = container + .readiness_probe + .expect("the broker kafka container must have a readinessProbe"); + let exec = readiness_probe + .exec + .expect("the readinessProbe must be an exec check"); + let command = exec.command.expect("exec has a command"); + assert_eq!(command[0], "/stackable/kcat"); + } + + #[test] + fn broker_pods_have_no_kcat_prober_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 == "kcat-prober"), + "expected no separate kcat-prober container, got: {:?}", + containers.iter().map(|c| &c.name).collect::>() + ); + } + + #[test] + fn controller_kafka_container_has_a_startup_probe() { + let cluster = kraft_mode_cluster(); + let container = controller_kafka_container(&cluster); + let client_port = cluster.cluster_config.kafka_security.client_port(); + + let startup_probe = container + .startup_probe + .expect("the controller kafka container must have a startupProbe"); + let tcp_socket = startup_probe + .tcp_socket + .expect("the startupProbe must be a tcpSocket check"); + assert_eq!(tcp_socket.port, IntOrString::Int(client_port.into())); + assert_eq!(startup_probe.timeout_seconds, Some(5)); + assert_eq!(startup_probe.period_seconds, Some(5)); + assert_eq!(startup_probe.failure_threshold, Some(60)); + } + + /// The liveness probe must check both TCP reachability (a genuinely dead/hung process must + /// still be restarted, same as before) and local Raft state, failing specifically on + /// `unattached` — see `controller_stuck_unattached_liveness_probe`'s doc comment for why + /// only that state, not any non-healthy state, is treated as restart-worthy. + #[test] + fn controller_kafka_container_liveness_probe_checks_tcp_and_stuck_unattached_state() { + let cluster = kraft_mode_cluster(); + let container = controller_kafka_container(&cluster); + let client_port = cluster.cluster_config.kafka_security.client_port(); + + let liveness_probe = container + .liveness_probe + .expect("the controller kafka container must have a livenessProbe"); + let exec = liveness_probe + .exec + .expect("the livenessProbe must be an exec check, not a bare tcpSocket check"); + let command = exec.command.expect("exec has a command"); + let script = command.last().expect("the exec command has a script arg"); + + assert!( + script.contains(&format!("/dev/tcp/localhost/{client_port}")), + "expected a TCP reachability check against the controller's own port, script was: {script}" + ); + assert!( + script.contains(r#"[ "$state" != "unattached" ]"#), + "expected the check to fail specifically (and only) on the unattached state, \ + script was: {script}" + ); + // Must not fail merely for being non-healthy in some *other* way (e.g. `candidate` or + // `observer`) - only `unattached` is the specific, restart-fixable symptom. + assert!(!script.contains("leader|follower")); + + assert_eq!(liveness_probe.timeout_seconds, Some(10)); + assert_eq!(liveness_probe.period_seconds, Some(30)); + assert_eq!(liveness_probe.failure_threshold, Some(20)); + } + + #[test] + fn controller_kafka_container_readiness_probe_checks_raft_state() { + let cluster = kraft_mode_cluster(); + let container = controller_kafka_container(&cluster); + + let readiness_probe = container.readiness_probe.expect("readiness probe is set"); + let exec = readiness_probe + .exec + .expect("readiness probe is an exec check"); + let command = exec.command.expect("exec has a command"); + assert_eq!( + command, + vec![ + "bash".to_string(), + "-c".to_string(), + "curl -s localhost:9606/metrics | grep -E 'kafka_server_raft_metrics_current_state\\{state=\"(leader|follower|voted)\",?\\}'".to_string(), + ] + ); + assert_eq!(readiness_probe.timeout_seconds, Some(10)); + assert_eq!(readiness_probe.period_seconds, Some(10)); + assert_eq!(readiness_probe.failure_threshold, Some(6)); } } diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index e36191c10..46592856c 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -49,8 +49,8 @@ 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"; +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,18 +221,46 @@ 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( security: &ValidatedKafkaSecurity, pod_builder: &mut PodBuilder, - cb_kcat_prober: &mut ContainerBuilder, cb_kafka: &mut ContainerBuilder, requested_secret_lifetime: &Duration, ) -> Result<(), Error> { // add tls (server or client authentication volumes) if required if let Some(tls_server_secret_class) = tls_secret_class(security) { - // We have to mount tls pem files for kcat (the mount can be used directly) + // We have to mount tls pem files for kcat's readiness-probe command (the mount can be + // used directly) pod_builder .add_volume(create_kcat_tls_volume( STACKABLE_TLS_KCAT_VOLUME_NAME, @@ -240,7 +268,7 @@ pub fn add_broker_volume_and_volume_mounts( requested_secret_lifetime, )?) .context(AddVolumeSnafu)?; - cb_kcat_prober + cb_kafka .add_volume_mount(STACKABLE_TLS_KCAT_VOLUME_NAME, STACKABLE_TLS_KCAT_DIR) .context(AddVolumeMountSnafu)?; // Keystores fore the kafka container @@ -651,7 +679,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 +752,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 +916,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/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index ec747fd71..f46d87a73 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -101,6 +101,14 @@ pub enum Error { "the Vector aggregator discovery ConfigMap name is required when the Vector agent is enabled" ))] MissingVectorAggregatorConfigMapName, + + #[snafu(display( + "at least one KRaft controller replica is required while any broker replicas are \ + configured; a KRaft cluster with zero controllers has no metadata quorum for its \ + brokers to use. Scale brokers to 0 as well (or use `clusterOperation.stopped`) for a \ + coordinated full stop" + ))] + NoKraftControllerReplicas, } /// Validated logging configuration for a Kafka role group's Kafka and (optional) Vector @@ -261,6 +269,10 @@ pub fn validate( ); role_group_configs.insert(KafkaRole::Broker, broker_groups); + let metadata_manager = kafka + .effective_metadata_manager() + .context(InvalidMetadataManagerSnafu)?; + // Controllers are optional: ZooKeeper-mode clusters have none, in which case they are simply // absent from both maps and not reconciled. if let Some(controller_role) = kafka.spec.controllers.as_ref() { @@ -273,6 +285,27 @@ pub fn validate( validate_controller_logging, &vector_aggregator_config_map_name, )?; + + // A KRaft cluster with zero controller replicas *and running brokers* is a broken + // half-state. Reject that combination here. + // + // Controllers *and* brokers at zero together is not rejected: that is exactly what + // `clusterOperation.stopped` already does today. + let controller_replicas: u16 = controller_groups + .values() + .map(|rg| rg.replicas.unwrap_or(1)) + .sum(); + let broker_replicas: u16 = role_group_configs[&KafkaRole::Broker] + .values() + .map(|rg| rg.replicas.unwrap_or(1)) + .sum(); + if metadata_manager == crate::crd::MetadataManager::KRaft + && controller_replicas == 0 + && broker_replicas > 0 + { + return NoKraftControllerReplicasSnafu.fail(); + } + role_configs.insert( KafkaRole::Controller, ValidatedRoleConfig { @@ -282,10 +315,6 @@ pub fn validate( role_group_configs.insert(KafkaRole::Controller, controller_groups); } - let metadata_manager = kafka - .effective_metadata_manager() - .context(InvalidMetadataManagerSnafu)?; - let name = get_cluster_name(kafka).context(ResolveClusterNameSnafu)?; let namespace = get_namespace(kafka).context(ResolveNamespaceSnafu)?; let uid = get_uid(kafka).context(ResolveUidSnafu)?; @@ -409,7 +438,7 @@ mod tests { builder::pod::container::EnvVarSet, types::operator::RoleGroupName, }; - use super::{KAFKA_CLUSTER_ID, inject_cluster_id}; + use super::{Error, KAFKA_CLUSTER_ID, inject_cluster_id}; use crate::{ controller::test_support::{app_version_label, minimal_kafka, validated_cluster}, crd::role::KafkaRole, @@ -533,4 +562,157 @@ mod tests { let env = inject_cluster_id(EnvVarSet::new(), None); assert_eq!(cluster_id_value(&env), None); } + + /// Confirmed live: scaling a KRaft cluster's only controller role group down to 0 replicas + /// while brokers keep running used to pass validation and fail much later and much more + /// confusingly, as `NoKraftControllersFound` while building the *broker* role group's + /// ConfigMap (which also reads the controller quorum's pod descriptors, to render + /// `controller.quorum.bootstrap.servers`). Brokers with zero controllers have no metadata + /// quorum to talk to at all, so this combination must be rejected here, at validation time, + /// with a message that actually names the real problem. + #[test] + fn kraft_mode_rejects_zero_controller_replicas_while_brokers_are_running() { + 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: 0 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + + let result = crate::controller::test_support::validate_err(&kafka); + let Err(error) = result else { + panic!( + "validate should reject zero controller replicas while brokers are running in KRaft mode" + ); + }; + + assert!( + matches!(error, Error::NoKraftControllerReplicas), + "expected NoKraftControllerReplicas, got: {error:?}" + ); + } + + /// The same zero-sum check, but split across two controller role groups (0 + 0): neither + /// group alone looks suspicious, only their sum does. + #[test] + fn kraft_mode_rejects_zero_controller_replicas_summed_across_role_groups() { + 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: + a: + replicas: 0 + b: + replicas: 0 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + + let result = crate::controller::test_support::validate_err(&kafka); + let Err(error) = result else { + panic!( + "validate should reject zero controller replicas while brokers are running in KRaft mode" + ); + }; + + assert!( + matches!(error, Error::NoKraftControllerReplicas), + "expected NoKraftControllerReplicas, got: {error:?}" + ); + } + + /// Controllers *and* brokers at zero together is not rejected: that is exactly what + /// `clusterOperation.stopped` already does today, unconditionally, for every Stackable + /// operator, bypassing this check entirely -- a coordinated whole-cluster stop is already a + /// supported shape, not a broken half-state the way controllers-only-at-zero is. + #[test] + fn kraft_mode_allows_controllers_and_brokers_at_zero_together() { + 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: 0 + brokers: + roleGroups: + default: + replicas: 0 + "#, + ); + + let _cluster = validated_cluster(&kafka); + } + + /// A `replicas: 0` controller role group is fine in ZooKeeper mode: the check only applies + /// to KRaft, where controllers *are* the metadata quorum. + #[test] + fn zookeeper_mode_allows_zero_controller_replicas() { + 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: + zookeeperConfigMapName: zk-discovery + controllers: + roleGroups: + default: + replicas: 0 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + + let _cluster = validated_cluster(&kafka); + } } diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index c2d118256..952f8a6fb 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -338,42 +338,6 @@ pub struct KafkaPodDescriptor { pub client_port: Port, } -impl KafkaPodDescriptor { - /// Return the fully qualified domain name - /// Format: `...svc.` - pub fn fqdn(&self) -> String { - format!( - "{pod_name}.{service_name}.{namespace}.svc.{cluster_domain}", - pod_name = self.pod_name(), - service_name = self.role_group_service_name, - namespace = self.namespace, - cluster_domain = self.cluster_domain - ) - } - - pub fn pod_name(&self) -> String { - format!("{}-{}", self.role_group_statefulset_name, self.replica) - } - - /// Build the Kraft voter String - /// See: - /// Example: 0@controller-0:1234:0000000000-00000000000 - /// * 0 is the replica id - /// * 0000000000-00000000000 is the replica directory id (even though the used Uuid states to be type 4 it does not work) - /// See: - /// * controller-0 is the replica's host, - /// * 1234 is the replica's port. - // NOTE(@maltesander): Even though the used Uuid states to be type 4 it does not work... 0000000000-00000000000 works... - pub fn as_voter(&self) -> String { - format!( - "{node_id}@{fqdn}:{port}:0000000000-{node_id:0>11}", - node_id = self.node_id, - port = self.client_port, - fqdn = self.fqdn(), - ) - } -} - #[derive(Clone, Default, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct KafkaClusterStatus { diff --git a/rust/operator-binary/src/crd/role/broker.rs b/rust/operator-binary/src/crd/role/broker.rs index c1eafa342..3c9b70a13 100644 --- a/rust/operator-binary/src/crd/role/broker.rs +++ b/rust/operator-binary/src/crd/role/broker.rs @@ -31,7 +31,6 @@ use crate::crd::role::commons::{CommonConfig, Storage, StorageFragment}; #[strum(serialize_all = "kebab-case")] pub enum BrokerContainer { Vector, - KcatProber, Kafka, } diff --git a/tests/templates/kuttl/configuration/10-assert.yaml.j2 b/tests/templates/kuttl/configuration/10-assert.yaml.j2 index 3de5ea661..3d75ac23f 100644 --- a/tests/templates/kuttl/configuration/10-assert.yaml.j2 +++ b/tests/templates/kuttl/configuration/10-assert.yaml.j2 @@ -23,7 +23,6 @@ spec: cpu: 250m # value set in the rolegroup configuration memory: 3Gi - - name: kcat-prober {% if lookup('env', 'VECTOR_AGGREGATOR') %} - name: vector {% endif %} diff --git a/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 index 61968a8a9..0dce51c18 100644 --- a/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/60-assert.yaml.j2 @@ -1,10 +1,20 @@ -{% if not test_scenario['values']['kafka-kraft'].startswith("3.7") %} --- apiVersion: kuttl.dev/v1beta1 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 +23,7 @@ metadata: status: readyReplicas: 3 replicas: 3 +--- apiVersion: apps/v1 kind: StatefulSet metadata: @@ -20,4 +31,3 @@ metadata: status: readyReplicas: 5 replicas: 5 -{% endif %} diff --git a/tests/templates/kuttl/operations-kraft/60-scale-controller-up.yaml.j2 b/tests/templates/kuttl/operations-kraft/60-scale-controller-up.yaml.j2 index 3fdc5c4da..5ce9614a8 100644 --- a/tests/templates/kuttl/operations-kraft/60-scale-controller-up.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/60-scale-controller-up.yaml.j2 @@ -1,4 +1,3 @@ -{% if not test_scenario['values']['kafka-kraft'].startswith("3.7") %} --- apiVersion: kuttl.dev/v1beta1 kind: TestStep @@ -38,4 +37,3 @@ spec: clusterOperation: stopped: false reconciliationPaused: false -{% endif %} diff --git a/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 index cd8c8ae2d..b4de15cf6 100644 --- a/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/70-assert.yaml.j2 @@ -1,10 +1,20 @@ -{% if not test_scenario['values']['kafka-kraft'].startswith("3.7") %} --- apiVersion: kuttl.dev/v1beta1 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 +23,7 @@ metadata: status: readyReplicas: 3 replicas: 3 +--- apiVersion: apps/v1 kind: StatefulSet metadata: @@ -20,4 +31,3 @@ metadata: status: readyReplicas: 3 replicas: 3 -{% endif %} diff --git a/tests/templates/kuttl/operations-kraft/70-scale-controller-down.yaml.j2 b/tests/templates/kuttl/operations-kraft/70-scale-controller-down.yaml.j2 index a077213ba..a6ad4ec2a 100644 --- a/tests/templates/kuttl/operations-kraft/70-scale-controller-down.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/70-scale-controller-down.yaml.j2 @@ -1,4 +1,3 @@ -{% if not test_scenario['values']['kafka-kraft'].startswith("3.7") %} --- apiVersion: kuttl.dev/v1beta1 kind: TestStep @@ -38,4 +37,3 @@ spec: clusterOperation: stopped: false reconciliationPaused: false -{% endif %} diff --git a/tests/templates/kuttl/operations-kraft/80-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/80-assert.yaml.j2 index a1d7088f5..793f8aad1 100644 --- a/tests/templates/kuttl/operations-kraft/80-assert.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/80-assert.yaml.j2 @@ -1,4 +1,3 @@ -{% if not test_scenario['values']['kafka-kraft'].startswith("3.7") %} --- apiVersion: kuttl.dev/v1beta1 kind: TestAssert @@ -18,4 +17,3 @@ metadata: status: readyReplicas: 3 replicas: 3 -{% endif %} diff --git a/tests/templates/kuttl/operations-kraft/80-scale-broker-down.yaml.j2 b/tests/templates/kuttl/operations-kraft/80-scale-broker-down.yaml.j2 index d788a9c90..ee4cb139f 100644 --- a/tests/templates/kuttl/operations-kraft/80-scale-broker-down.yaml.j2 +++ b/tests/templates/kuttl/operations-kraft/80-scale-broker-down.yaml.j2 @@ -4,7 +4,6 @@ # The brokers must be deleted because otherwise they are left dangling until # the test timeouts and fails. # -{% if not test_scenario['values']['kafka-kraft'].startswith("3.7") %} --- apiVersion: kuttl.dev/v1beta1 kind: TestStep @@ -44,4 +43,3 @@ spec: clusterOperation: stopped: false reconciliationPaused: false -{% endif %} diff --git a/tests/templates/kuttl/operations-kraft/90-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/90-assert.yaml.j2 new file mode 100644 index 000000000..41e7d53da --- /dev/null +++ b/tests/templates/kuttl/operations-kraft/90-assert.yaml.j2 @@ -0,0 +1,18 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-default +status: + replicas: 0 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-controller-default +status: + replicas: 0 diff --git a/tests/templates/kuttl/operations-kraft/90-controller-shutdown.yaml.j2 b/tests/templates/kuttl/operations-kraft/90-controller-shutdown.yaml.j2 new file mode 100644 index 000000000..6ccff4ad9 --- /dev/null +++ b/tests/templates/kuttl/operations-kraft/90-controller-shutdown.yaml.j2 @@ -0,0 +1,45 @@ +# +# This is a test helper to ensure that all broker pods are deleted before +# the test namespace is terminated. +# The brokers must be deleted because otherwise they are left dangling until +# the test timeouts and fails. +# +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +timeout: 600 +--- +apiVersion: kafka.stackable.tech/v1alpha1 +kind: KafkaCluster +metadata: + name: test-kafka +spec: + image: +{% if test_scenario['values']['kafka-kraft'].find(",") > 0 %} + custom: "{{ test_scenario['values']['kafka-kraft'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['kafka-kraft'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['kafka-kraft'] }}" +{% endif %} +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + clusterConfig: + metadataManager: kraft + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + controllers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + roleGroups: + default: + replicas: 0 + brokers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + roleGroups: + default: + replicas: 0 + clusterOperation: + stopped: false + reconciliationPaused: false diff --git a/tests/templates/kuttl/operations-kraft/README.md b/tests/templates/kuttl/operations-kraft/README.md deleted file mode 100644 index 5c0fa86b3..000000000 --- a/tests/templates/kuttl/operations-kraft/README.md +++ /dev/null @@ -1,14 +0,0 @@ -Tests Kraft cluster operations: - -- Cluster stop/pause/restart -- Scale brokers up/down -- Scale controllers up/down - -Notes - -- Kafka 3.7 controllers do not scale at all. - The scaling test steps are disabled for this version. -- Scaling controllers from 3 -> 1 doesn't work. - 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. diff --git a/tests/templates/kuttl/smoke/33-assert.yaml.j2 b/tests/templates/kuttl/smoke/33-assert.yaml.j2 index 70678faa8..01d3817f9 100644 --- a/tests/templates/kuttl/smoke/33-assert.yaml.j2 +++ b/tests/templates/kuttl/smoke/33-assert.yaml.j2 @@ -64,14 +64,6 @@ spec: requests: cpu: 300m # From podOverrides memory: 2Gi - - name: kcat-prober - resources: - limits: - cpu: 200m - memory: 128Mi - requests: - cpu: 100m - memory: 128Mi {% if vector_enabled %} - name: vector env: diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 3cb4633fe..9b47243af 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -146,12 +146,7 @@ suites: - name: nightly patch: - dimensions: - - name: kafka - expr: last - - name: zookeeper - expr: last - - name: upgrade_old - expr: last + - expr: last - name: smoke-latest select: - smoke