diff --git a/.gitignore b/.gitignore index 411cc85da..17cb4062d 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,5 @@ tilt_options.json .envrc .DS_Store + +.worktrees/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 207a0c37f..454e23a64 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 prevents the + restart controller from restarting all pods in the quorum when a new one is added/deleted. + The controller `StatefulSet` is now scaled using `OrderedReady` 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]). @@ -34,12 +45,24 @@ All notable changes to this project will be documented in this file. 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]). - The operator now watches all resources that it creates and early-exits the reconcile action when the cluster is marked for deletion ([#1014]). +- A KRaft cluster without a `controllers` role is now rejected during validation. + Previously, the operator would create a cluster consisting only of `brokers` which would never + become healthy ([#1010]). + +### 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 [#1014]: https://github.com/stackabletech/kafka-operator/pull/1014 [#1017]: https://github.com/stackabletech/kafka-operator/pull/1017 diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 455188c97..766686352 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -85,33 +85,29 @@ 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. + On pod restart or termination, the controller is removed from the voter set by the `preStop` hook on the `kafka` container. +* 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. +* `controller.quorum.bootstrap.servers` 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 quorum is created once by the controller with the lowest `node.id` using `--standalone`. If this controller + loses it's PVC, a new conflicting quorum is created on restart. +* A Controller that loses its persistent volume is not re-admitted to the voter set automatically, because it + returns with a new directory ID while the quorum still lists the old one. -== Troubleshooting +==== Prevention -=== Cluster does not start - -Check that at least a quorum (majority) of controllers are reachable. - -=== Frequent leader elections - -Likely caused by controller resource starvation or unstable Kubernetes scheduling. - -=== Migration issues (ZooKeeper to KRaft) - -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 - -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). +* Use a StorageClass backed by network-attached storage, so that a node failure does not imply volume loss. +* Do not delete Controller PVCs as part of routine maintenance. == 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 e077db3c7..21e1d8de4 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -611,6 +611,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. Used for tests asserting on a specific validation failure. + pub fn validate_err( + kafka: &v1alpha1::KafkaCluster, + ) -> Result { validate( kafka, DereferencedObjects { @@ -621,7 +629,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..dbf609b9c 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,14 @@ 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"); +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 +69,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 +88,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 +100,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 +115,1033 @@ 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 -} +/// Selects the quorum format flag for the given controller. +/// +/// The controller with lowest `node_id` starts with `--standalone` while all others +/// start with `--no-initial-controllers` and are added later to the voter list +/// by the `quorum-manager`. +/// +/// Known limitation: If the controller with the lowest `node_id` loses it's PVC it will +/// create a new conflicting quorum upon restart. +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, + ) } -fn initial_controllers_command( - controller_descriptors: &[KafkaPodDescriptor], - product_version: &str, +/// How often the sidecar's admission loop polls. +const QUORUM_MANAGER_POLL_INTERVAL_SECONDS: u32 = 10; + +/// Consecutive healthy polls this controller must report before it will join a quorum that +/// currently has a *single* voter. +const QUORUM_MANAGER_STABILITY_REQUIRED_POLLS: u32 = 4; + +/// How long an existing voter may go without fetching from the leader before the sidecar +/// treats the quorum as degraded and defers changing its membership. +const QUORUM_MANAGER_VOTER_STALE_FETCH_SECONDS: u32 = 30; + +const CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT: &str = + include_str!("scripts/controller-quorum-manager-loop.sh"); + +/// The sidecar's main-loop command: while this controller's local Raft state is `observer`, +/// admit it into the quorum's voter set once that is safe. +pub fn quorum_manager_container_command() -> String { + format!( + r#" + set -uo pipefail + + {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 + QUORUM_CLI={binary} + ADMIN_CLIENT_CONFIG={admin_client_config} + ADD_CONTROLLER_CONFIG={add_controller_config} + METRICS_URL=localhost:{metrics_port}/metrics + CLI_TIMEOUT_SECONDS={cli_timeout} + CLI_KILL_AFTER_SECONDS={cli_kill_after} + POLL_INTERVAL_SECONDS={poll_interval} + STABILITY_REQUIRED_POLLS={stability_polls} + VOTER_STALE_FETCH_SECONDS={stale_fetch} +{script} + 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, + poll_interval = QUORUM_MANAGER_POLL_INTERVAL_SECONDS, + stability_polls = QUORUM_MANAGER_STABILITY_REQUIRED_POLLS, + stale_fetch = QUORUM_MANAGER_VOTER_STALE_FETCH_SECONDS, + script = strip_shell_comments(CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT), + ) +} + +/// 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| { + let secs = timeout.as_secs(); + secs.saturating_sub(PRE_STOP_RESERVED_FOR_KAFKA_SHUTDOWN_SECONDS) + .clamp(PRE_STOP_MIN_DEADLINE_SECONDS, PRE_STOP_MAX_DEADLINE_SECONDS) + .min(secs) // never outlive the grace period itself + }) + .unwrap_or(PRE_STOP_MIN_DEADLINE_SECONDS) +} + +/// Pause (seconds) between two voter-removal attempts inside the `preStop` script's retry +/// loop. +const PRE_STOP_RETRY_INTERVAL_SECONDS: u32 = 2; + +const CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT: &str = + include_str!("scripts/controller-remove-self-pre-stop.sh"); + +/// Drops whole-line `#` comments, and the blank runs they leave behind, from a shell script. +fn strip_shell_comments(script: &str) -> String { + let mut kept: Vec<&str> = Vec::new(); + + for line in script + .lines() + .filter(|line| !line.trim_start().starts_with('#')) + { + let previous_is_blank = kept.last().is_none_or(|line: &&str| line.trim().is_empty()); + if line.trim().is_empty() && previous_is_blank { + continue; + } + kept.push(line); + } + while kept.last().is_some_and(|line| line.trim().is_empty()) { + kept.pop(); + } + + kept.join("\n") +} + +/// The `kafka` container's own `preStop` command: before this controller pod terminates, +/// remove it from the KRaft voter set, unless that would remove the *last* remaining voter. +/// +/// This only assembles the preamble that feeds +/// [`CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT`] its inputs; the logic, and the reasoning behind +/// it, lives in that script (whose maintenance comments [`strip_shell_comments`] drops on +/// the way in). +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), - ), + formatdoc! {" + set -uo pipefail + {derive_pod_index} + [ -n \"$POD_INDEX\" ] || exit 0 + {export_replica_id} + {extract_bootstrap_servers} + QUORUM_CLI={binary} + ADMIN_CLIENT_CONFIG={config} + CLI_TIMEOUT_SECONDS={cli_timeout} + CLI_KILL_AFTER_SECONDS={cli_kill_after} + REMOVAL_DEADLINE_SECONDS={deadline_seconds} + RETRY_INTERVAL_SECONDS={retry_interval} + {script}", + derive_pod_index = DERIVE_POD_INDEX, + export_replica_id = EXPORT_REPLICA_ID, + extract_bootstrap_servers = extract_bootstrap_servers_command(), + binary = KAFKA_METADATA_QUORUM_BINARY, + config = ADMIN_CLIENT_PROPERTIES_PATH, + cli_timeout = CLI_CALL_TIMEOUT_SECONDS, + cli_kill_after = CLI_CALL_KILL_AFTER_SECONDS, + deadline_seconds = pre_stop_deadline_seconds(graceful_shutdown_timeout), + retry_interval = PRE_STOP_RETRY_INTERVAL_SECONDS, + script = strip_shell_comments(CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT), } } #[cfg(test)] mod tests { + use std::{fs, os::unix::fs::PermissionsExt, process::Command}; + + use indoc::indoc; + 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"#)); + } + + /// 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("trap 'handle_term_signal' TERM")); + // The poll interval is slept in the background and `wait`ed on, so bash can run the + // trap immediately instead of only after a foreground `sleep` returns. + assert!(command.contains(r#"sleep "$POLL_INTERVAL_SECONDS" &"#)); + assert!(command.contains("wait $!")); + assert!(command.contains(&format!( + "POLL_INTERVAL_SECONDS={QUORUM_MANAGER_POLL_INTERVAL_SECONDS}" + ))); + 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. + // Join line continuations first: the invocation is spread over several lines. + let joined = command.replace("\\\n", " "); + let add_controller_line = joined + .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. + /// + /// `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 fails with `node.id not found in configuration file`. See + /// [`ADD_CONTROLLER_PROPERTIES_PATH`] for why this is a merged file (in this specific + /// order) rather than `controller.properties` outright (that file has no bare + /// `ssl.*`/`security.protocol`, so the AdminClient cannot 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" + )); + // The merged file is what `add-controller` — and only `add-controller` — connects + // with; read-only `describe` calls keep using the plain admin-client config. + assert!(command.contains("ADD_CONTROLLER_CONFIG=/tmp/add-controller.properties")); + let joined = command.replace("\\\n", " "); + let add_controller_line = joined + .lines() + .find(|line| line.contains("add-controller") && line.contains("--command-config")) + .expect("the add-controller invocation is present"); + assert!( + add_controller_line.contains(r#"--command-config "$ADD_CONTROLLER_CONFIG""#), + "add-controller must use the merged config, line was: {add_controller_line}" + ); + } + + /// Stands in for `kafka-metadata-quorum.sh` in [`run_pre_stop`]: records every + /// invocation, answers `describe --replication` from `$DESCRIBE_OUTPUT`, and fails the + /// first `$REMOVE_CONTROLLER_FAILURES` `remove-controller` calls before succeeding. + const STUB_QUORUM_CLI: &str = indoc! {r#" + #!/usr/bin/env bash + set -u + echo "$*" >> "$CALL_LOG" + for arg in "$@"; do + case "$arg" in + describe) + cat "$DESCRIBE_OUTPUT" + exit 0 + ;; + remove-controller) + attempts=$(( $(cat "$ATTEMPTS") + 1 )) + echo "$attempts" > "$ATTEMPTS" + if [ "$attempts" -le "$REMOVE_CONTROLLER_FAILURES" ]; then + exit 1 + fi + exit 0 + ;; + esac + done + exit 0 + "#}; + + /// What one execution of [`CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT`] did. + struct PreStopRun { + exit_code: Option, + stdout: String, + stderr: String, + /// The stub CLI's argument list, one entry per invocation, in call order. + cli_calls: Vec, + } + + impl PreStopRun { + /// The recorded invocations of the given subcommand. + fn calls_of(&self, subcommand: &str) -> Vec<&String> { + self.cli_calls + .iter() + .filter(|call| call.contains(subcommand)) + .collect() + } + } + + /// The inputs of one [`run_pre_stop`] scenario. [`Default`] describes an unreachable + /// quorum, so each test only spells out what it actually cares about. + struct PreStopScenario<'a> { + /// Unique per test — names this run's scratch directory. + name: &'a str, + /// What the stub prints for `describe --replication`. Empty output stands for an + /// unreachable quorum (or a call `timeout` killed). + describe: String, + /// The script's total retry budget, i.e. what [`pre_stop_deadline_seconds`] would + /// produce in production. + deadline_seconds: u32, + } + + impl Default for PreStopScenario<'_> { + fn default() -> Self { + Self { + name: "unnamed", + describe: String::new(), + deadline_seconds: 2, + } + } + } + + /// Mimics one `kafka-metadata-quorum.sh describe --replication` table: a header row, then + /// one `NodeId DirectoryId ... Status` row per replica. + fn describe_output(replicas: &[(u32, &str, &str)]) -> String { + let mut output = "NodeId\tDirectoryId\tLogEndOffset\tLag\tLastFetchTimestamp\tLastCaughtUpTimestamp\tStatus\n".to_string(); + for (node_id, directory_id, status) in replicas { + output.push_str(&format!( + "{node_id}\t{directory_id}\t100\t0\t1758000000\t1758000000\t{status}\n" + )); + } + output + } + + /// Executes the real `preStop` script in bash against a stub `kafka-metadata-quorum.sh`, + /// in the comment-stripped form that actually ships in the pod template. + /// + /// This exercises the script's own contract — the env inputs its header documents — not + /// the operator-generated preamble that supplies them in production; the two are kept in + /// sync by [`generated_pre_stop_command_defines_every_input_the_script_requires`]. + fn run_pre_stop(scenario: PreStopScenario) -> PreStopRun { + let dir = std::env::temp_dir().join(format!( + "kafka-operator-pre-stop-{name}-{pid}", + name = scenario.name, + pid = std::process::id() + )); + let _ = fs::remove_dir_all(&dir); + fs::create_dir_all(&dir).expect("the scratch directory can be created"); + + let stub_cli = dir.join("kafka-metadata-quorum.sh"); + fs::write(&stub_cli, STUB_QUORUM_CLI).expect("the stub CLI can be written"); + fs::set_permissions(&stub_cli, fs::Permissions::from_mode(0o755)) + .expect("the stub CLI can be made executable"); + + let describe_output_file = dir.join("describe-output"); + fs::write(&describe_output_file, &scenario.describe) + .expect("the stub's describe output can be written"); + let call_log = dir.join("cli-calls"); + fs::write(&call_log, "").expect("the stub's call log can be created"); + let attempts = dir.join("remove-controller-attempts"); + fs::write(&attempts, "0").expect("the stub's attempt counter can be created"); + + let mut command = Command::new("bash"); + command + .arg("-c") + .arg(strip_shell_comments(CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT)) + // The script's own inputs. + .env( + "BOOTSTRAP_SERVERS", + "kafka-controller-default-headless.default.svc.cluster.local:9093", + ) + .env("QUORUM_CLI", &stub_cli) + .env("ADMIN_CLIENT_CONFIG", dir.join("admin-client.properties")) + .env("CLI_TIMEOUT_SECONDS", "5") + .env("CLI_KILL_AFTER_SECONDS", "1") + .env( + "REMOVAL_DEADLINE_SECONDS", + scenario.deadline_seconds.to_string(), + ) + .env("RETRY_INTERVAL_SECONDS", "1") + // Read by the stub CLI, not by the script under test. + .env("DESCRIBE_OUTPUT", &describe_output_file) + .env("CALL_LOG", &call_log) + .env("ATTEMPTS", &attempts) + .env("REMOVE_CONTROLLER_FAILURES", "0") + .env("REPLICA_ID", "1"); + + let output = command + .output() + .expect("bash is available to run the preStop script"); + let cli_calls = fs::read_to_string(&call_log) + .expect("the stub's call log can be read") + .lines() + .map(str::to_string) + .collect(); + + let run = PreStopRun { + exit_code: output.status.code(), + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + cli_calls, + }; + fs::remove_dir_all(&dir).expect("the scratch directory can be removed"); + run + } + + /// The happy path: with other voters left behind, the departing controller removes itself, + /// passing both its node id and the directory id it read out of the quorum's own + /// `describe` output (`remove-controller` needs both). + #[test] + fn pre_stop_removes_self_while_other_voters_remain() { + let run = run_pre_stop(PreStopScenario { + name: "removes-self", + describe: describe_output(&[ + (1, "dir-1", "Leader"), + (2, "dir-2", "Follower"), + (3, "dir-3", "Follower"), + ]), + ..Default::default() + }); + + assert_eq!(run.exit_code, Some(0), "stderr was: {}", run.stderr); + assert!( + run.stdout + .contains("Removing self (node 1, directory dir-1)") + ); + let removals = run.calls_of("remove-controller"); + assert_eq!(removals.len(), 1, "cli calls were: {:?}", run.cli_calls); + assert!( + removals[0].contains("--controller-id 1 --controller-directory-id dir-1"), + "removal call was: {}", + removals[0] + ); + } + + /// Removing the last voter would break the next cluster restart (it would reformat the + /// Raft metadata), so it must never happen — and it must give up immediately rather than + /// retry: no peer can add a voter on this pod's behalf while it is terminating, so the + /// answer can never change. Observers are not voters, so the one in this quorum must not + /// be counted as the spare that would make the removal look safe. + #[test] + fn pre_stop_never_removes_the_last_voter_and_gives_up_immediately() { + let run = run_pre_stop(PreStopScenario { + name: "last-voter", + describe: describe_output(&[(1, "dir-1", "Leader"), (2, "dir-2", "Observer")]), + deadline_seconds: 10, + }); + + assert_eq!(run.exit_code, Some(0), "stderr was: {}", run.stderr); + assert!(run.stdout.contains("Removing self would leave zero voters")); + assert!( + run.calls_of("remove-controller").is_empty(), + "cli calls were: {:?}", + run.cli_calls + ); + assert_eq!( + run.calls_of("describe").len(), + 1, + "the last-voter case must not retry until the deadline, cli calls were: {:?}", + run.cli_calls + ); + } + + /// An unreachable quorum is retried for the whole budget and then reported loudly + /// (`ERROR:`, so it is greppable and alertable): the on-disk voter set may now list a pod + /// that is gone, which is what can strand a later restart-from-zero. The hook still exits + /// 0 — a failed removal must never be why a pod fails to terminate. + #[test] + fn pre_stop_retries_an_unreachable_quorum_then_reports_loudly() { + let run = run_pre_stop(PreStopScenario { + name: "unreachable", + deadline_seconds: 3, + ..Default::default() + }); + + assert_eq!(run.exit_code, Some(0), "stderr was: {}", run.stderr); + assert!( + run.calls_of("describe").len() >= 2, + "an unreachable quorum must be retried, cli calls were: {:?}", + run.cli_calls + ); + assert!( + run.calls_of("remove-controller").is_empty(), + "cli calls were: {:?}", + run.cli_calls + ); + assert!( + run.stdout + .contains("ERROR: could not remove self (node 1) from the voter set"), + "stdout was: {}", + run.stdout + ); + } + + /// The script and the preamble feeding it live in different files, so nothing but this + /// test keeps the two in sync: every input the script declares mandatory has to be + /// assigned by the generated command. + #[test] + fn generated_pre_stop_command_defines_every_input_the_script_requires() { + let command = controller_remove_self_pre_stop_command(None); + let required_inputs: Vec<&str> = CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT + .lines() + .filter_map(|line| line.trim().strip_prefix(r#": "${"#)) + .filter_map(|declaration| declaration.split(":?").next()) + .collect(); + + assert!( + !required_inputs.is_empty(), + "the script is expected to declare its mandatory inputs as `: \"${{NAME:?...}}\"`" + ); + for input in required_inputs { + assert!( + command.contains(&format!("{input}=")), + "the generated preStop preamble must define `{input}`, which the script \ + declares mandatory; command was: {command}" + ); + } + } + + /// 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. + assert_eq!(pre_stop_deadline_seconds(Some(Duration::from_secs(10))), 10); + + // 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 + ); + + // ...and the generated script is handed that value, rather than a literal. + let command = + controller_remove_self_pre_stop_command(Some(Duration::from_minutes_unchecked(30))); + assert!(command.contains(&format!( + "REMOVAL_DEADLINE_SECONDS={PRE_STOP_MAX_DEADLINE_SECONDS}" + ))); + } + + /// `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 (`timeout 3 bash -c 'trap + /// "" TERM; sleep 30'` takes the full 30s, while `timeout --kill-after=2 3 ...` is + /// 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 the kind of thing a JVM AdminClient hangs on well past + /// its own `timeout` wrapper, delaying pod termination by minutes. + #[test] + fn every_cli_call_has_a_kill_after_so_timeout_is_actually_enforced() { + let container_command = quorum_manager_container_command(); + + // Both scripts are handed the CLI path by their operator-generated preamble and + // then refer to it by variable, so that assignment is the only place the literal + // path may appear in the rendered command. + for line in container_command + .lines() + .filter(|line| line.contains(KAFKA_METADATA_QUORUM_BINARY)) + { + assert!( + line.trim().starts_with("QUORUM_CLI="), + "the CLI path must only appear as the QUORUM_CLI assignment, so that every \ + actual invocation goes through the `timeout --kill-after=` wrappers checked \ + below — offending line: {line}" + ); + } + + for (script, cli_invocation) in [ + (CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT, r#""$QUORUM_CLI""#), + (CONTROLLER_REMOVE_SELF_PRE_STOP_SCRIPT, r#""$QUORUM_CLI""#), + ] { + // Join line continuations first: an invocation may well be spread over two lines. + let script = script.replace("\\\n", " "); + let invocations: Vec<&str> = script + .lines() + .filter(|line| line.contains(cli_invocation)) + .collect(); + + assert!( + !invocations.is_empty(), + "expected at least one `{cli_invocation}` invocation to check" + ); + for line in invocations { + 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}" + ); + } + } + } + + /// Stands in for `curl` in [`run_quorum_manager_loop`]: prints a metrics body reporting + /// `$METRIC_STATE`. + const STUB_METRICS_CURL: &str = indoc! {r#" + #!/usr/bin/env bash + set -u + echo "kafka_server_raft_metrics_current_state{state=\"$METRIC_STATE\",}" + exit 0 + "#}; + + /// One `describe --replication` table whose voters last fetched `fetch_age_seconds` ago. + fn quorum_describe_output(replicas: &[(u32, &str, &str)], fetch_age_seconds: u64) -> String { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("the clock is after the epoch") + .as_millis() as u64; + let fetched_ms = now_ms - fetch_age_seconds * 1000; + + let mut output = "NodeId\tDirectoryId\tLogEndOffset\tLag\tLastFetchTimestamp\tLastCaughtUpTimestamp\tStatus\n".to_string(); + for (node_id, directory_id, status) in replicas { + output.push_str(&format!( + "{node_id}\t{directory_id}\t100\t0\t{fetched_ms}\t{fetched_ms}\t{status}\n" + )); + } + output + } + + /// The inputs of one [`run_quorum_manager_loop`] scenario. + struct QuorumManagerScenario<'a> { + /// Unique per test — names this run's scratch directory. + name: &'a str, + /// What the stub CLI prints for `describe --replication`. Empty stands for an + /// unreachable quorum. + describe: String, + /// This controller's own Raft state, as its metrics endpoint reports it. + metric_state: &'a str, + /// Consecutive healthy polls required before joining a single-voter quorum. + stability_required_polls: u32, + /// How long the loop is left running, in seconds. + run_for_seconds: u32, + } + + impl Default for QuorumManagerScenario<'_> { + fn default() -> Self { + Self { + name: "unnamed", + describe: String::new(), + metric_state: "observer", + stability_required_polls: 3, + run_for_seconds: 2, + } + } + } + + /// What one bounded execution of [`CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT`] did. + struct QuorumManagerRun { + stdout: String, + cli_calls: Vec, + } + + impl QuorumManagerRun { + fn calls_of(&self, subcommand: &str) -> Vec<&String> { + self.cli_calls + .iter() + .filter(|call| call.contains(subcommand)) + .collect() + } + } + + /// Runs the real admission loop in bash against stub `kafka-metadata-quorum.sh` and + /// `curl` binaries, for a bounded time, in the comment-stripped form that ships. + /// + /// The loop never terminates on its own, so it is killed once `run_for_seconds` elapse; + /// with a one-second poll interval that is `run_for_seconds` iterations, give or take. + fn run_quorum_manager_loop(scenario: QuorumManagerScenario) -> QuorumManagerRun { + let dir = std::env::temp_dir().join(format!( + "kafka-operator-quorum-manager-{name}-{pid}", + name = scenario.name, + pid = std::process::id() + )); + let _ = fs::remove_dir_all(&dir); + fs::create_dir_all(&dir).expect("the scratch directory can be created"); + + let stub_cli = dir.join("kafka-metadata-quorum.sh"); + fs::write(&stub_cli, STUB_QUORUM_CLI).expect("the stub CLI can be written"); + fs::set_permissions(&stub_cli, fs::Permissions::from_mode(0o755)) + .expect("the stub CLI can be made executable"); + + // The script calls `curl` by bare name, so the stub is found via PATH. + let stub_curl = dir.join("curl"); + fs::write(&stub_curl, STUB_METRICS_CURL).expect("the stub curl can be written"); + fs::set_permissions(&stub_curl, fs::Permissions::from_mode(0o755)) + .expect("the stub curl can be made executable"); + + let describe_output_file = dir.join("describe-output"); + fs::write(&describe_output_file, &scenario.describe) + .expect("the stub's describe output can be written"); + let call_log = dir.join("cli-calls"); + fs::write(&call_log, "").expect("the stub's call log can be created"); + let attempts = dir.join("remove-controller-attempts"); + fs::write(&attempts, "0").expect("the stub's attempt counter can be created"); + + let path = format!( + "{stub_dir}:{existing}", + stub_dir = dir.display(), + existing = std::env::var("PATH").unwrap_or_default() + ); + + let mut command = Command::new("timeout"); + command + .arg(scenario.run_for_seconds.to_string()) + .arg("bash") + .arg("-c") + .arg(strip_shell_comments(CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT)) + .env("PATH", path) + .env("REPLICA_ID", "3") + .env( + "BOOTSTRAP_SERVERS", + "kafka-controller-default-headless.default.svc.cluster.local:9093", + ) + .env("QUORUM_CLI", &stub_cli) + .env("ADMIN_CLIENT_CONFIG", dir.join("admin-client.properties")) + .env( + "ADD_CONTROLLER_CONFIG", + dir.join("add-controller.properties"), + ) + .env("METRICS_URL", "localhost:9606/metrics") + .env("CLI_TIMEOUT_SECONDS", "5") + .env("CLI_KILL_AFTER_SECONDS", "1") + .env("POLL_INTERVAL_SECONDS", "1") + .env( + "STABILITY_REQUIRED_POLLS", + scenario.stability_required_polls.to_string(), + ) + .env("VOTER_STALE_FETCH_SECONDS", "30") + // Read by the stubs, not by the script under test. + .env("DESCRIBE_OUTPUT", &describe_output_file) + .env("CALL_LOG", &call_log) + .env("ATTEMPTS", &attempts) + .env("REMOVE_CONTROLLER_FAILURES", "0") + .env("METRIC_STATE", scenario.metric_state); + + let output = command.output().expect("bash is available to run the loop"); + let cli_calls = fs::read_to_string(&call_log) + .expect("the stub's call log can be read") + .lines() + .map(str::to_string) + .collect(); + + let run = QuorumManagerRun { + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + cli_calls, + }; + fs::remove_dir_all(&dir).expect("the scratch directory can be removed"); + run + } + + /// Joining the only existing voter makes both nodes load-bearing, so a controller that + /// has only just appeared must not be admitted yet. + #[test] + fn quorum_manager_defers_joining_a_single_voter_until_it_has_proven_stable() { + let run = run_quorum_manager_loop(QuorumManagerScenario { + name: "single-voter-probation", + describe: quorum_describe_output(&[(1, "dir-1", "Leader")], 1), + stability_required_polls: 10, + run_for_seconds: 3, + ..Default::default() + }); + + assert!( + run.calls_of("add-controller").is_empty(), + "a fresh controller must not join a single-voter quorum, calls: {:?}", + run.cli_calls + ); + assert!( + run.stdout.contains("proving stability before joining it"), + "stdout was: {}", + run.stdout + ); + } + + /// ...but it is admitted once the streak is met, otherwise the cluster could never grow. + #[test] + fn quorum_manager_joins_a_single_voter_once_stable() { + let run = run_quorum_manager_loop(QuorumManagerScenario { + name: "single-voter-admitted", + describe: quorum_describe_output(&[(1, "dir-1", "Leader")], 1), + stability_required_polls: 2, + run_for_seconds: 5, + ..Default::default() + }); + + assert!( + !run.calls_of("add-controller").is_empty(), + "a controller that stayed healthy must eventually join, stdout: {}", + run.stdout + ); + } + + /// Two voters is the fragile size — majority 2, so no failure is tolerated. Getting to + /// three is urgent, so this step is not delayed by the probation. + #[test] + fn quorum_manager_joins_a_two_voter_quorum_without_waiting() { + let run = run_quorum_manager_loop(QuorumManagerScenario { + name: "two-voter-immediate", + describe: quorum_describe_output( + &[(1, "dir-1", "Leader"), (2, "dir-2", "Follower")], + 1, + ), + stability_required_polls: 100, + run_for_seconds: 2, + ..Default::default() + }); + + assert!( + !run.calls_of("add-controller").is_empty(), + "leaving a two-voter quorum must not wait on probation, stdout: {}", + run.stdout + ); + } + + /// Never change the membership of a quorum that is already struggling: a voter that has + /// stopped fetching means the next change could be the one that loses the majority. + #[test] + fn quorum_manager_defers_while_an_existing_voter_is_stale() { + let run = run_quorum_manager_loop(QuorumManagerScenario { + name: "degraded-quorum", + describe: quorum_describe_output( + &[(1, "dir-1", "Leader"), (2, "dir-2", "Follower")], + 600, + ), + run_for_seconds: 3, + ..Default::default() + }); + + assert!( + run.calls_of("add-controller").is_empty(), + "a degraded quorum must not be perturbed, calls: {:?}", + run.cli_calls + ); + assert!( + run.stdout.contains("existing quorum is degraded"), + "stdout was: {}", + run.stdout + ); + } + + /// A controller that is already a voter has nothing to do. + #[test] + fn quorum_manager_does_nothing_when_already_a_voter() { + let run = run_quorum_manager_loop(QuorumManagerScenario { + name: "already-voter", + describe: quorum_describe_output(&[(1, "dir-1", "Leader")], 1), + metric_state: "follower", + run_for_seconds: 2, + ..Default::default() + }); + + assert!( + run.calls_of("add-controller").is_empty(), + "a voter must not re-add itself, calls: {:?}", + run.cli_calls + ); + assert!( + run.stdout.contains("Local Raft state is 'follower'"), + "stdout was: {}", + run.stdout + ); + } + + /// Builds a minimal [`KafkaPodDescriptor`] for the given role and replica. + 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`). + #[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\"" + )); + // No `--initial-controllers `, and no synthetic directory-id suffix. + assert!(!command.contains("--initial-controllers")); + assert!(!command.contains("0000000000-")); + } + + /// 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 7e2ebebea..cc2f1d919 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -52,7 +52,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> { @@ -75,12 +74,9 @@ pub fn add_kerberos_pod_config( ) .context(AddVolumeSnafu)?; - for cb in [cb_kafka, cb_kcat_prober] { - cb.add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR) - .expect( - "The mount paths are statically defined and there should be no duplicates.", - ); - } + cb_kafka + .add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR) + .expect("The mount paths are statically defined and there should be no duplicates."); } Ok(()) @@ -89,8 +85,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..1be5313e1 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -229,13 +229,16 @@ 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, + test_support::{ + bootstrap_listener, ingress_address, minimal_kafka, validated_cluster, + zookeeper_mode_cluster, + }, }, + crd::{STACKABLE_CONFIG_DIR, STACKABLE_DATA_DIR}, }; /// Sorted `metadata.name`s of the given resources, for order-independent assertions. @@ -278,6 +281,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(); @@ -332,9 +365,6 @@ mod tests { ); } - /// `build()` threads the bootstrap Listeners (fetched in the dereference step) through to the - /// discovery ConfigMap: once one carries an ingress address, the `KAFKA` entry names it. The - /// other tests run without bootstrap Listeners, where the entry is empty. #[test] fn build_writes_listener_addresses_to_the_discovery_configmap() { let mut cluster = kraft_mode_cluster(); @@ -365,6 +395,56 @@ 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. Without this mount the path + // doesn't exist in the sidecar and every attempt fails with "Unable to read + // meta.properties from /stackable/data/kraft". + 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:?}" + ); + } + /// 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..d607d5f31 100644 --- a/rust/operator-binary/src/controller/build/properties/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/mod.rs @@ -6,6 +6,8 @@ pub mod listener; pub mod product_logging; pub mod security_properties; +use std::collections::BTreeSet; + use crate::crd::{ KafkaPodDescriptor, role::{AnyConfig, KafkaRole}, @@ -25,6 +27,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 +63,27 @@ 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 .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::>() + .into_iter() + .collect() } #[cfg(test)] @@ -86,8 +102,107 @@ 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"); } + + fn pod_descriptor(role: KafkaRole, replica: u16, client_port: u16) -> KafkaPodDescriptor { + KafkaPodDescriptor { + namespace: "default".parse().expect("valid namespace name"), + role_group_statefulset_name: "kafka-controller-default" + .parse() + .expect("valid statefulset name"), + role_group_service_name: "kafka-controller-default-headless" + .parse() + .expect("valid service name"), + replica, + cluster_domain: stackable_operator::commons::networking::DomainName::try_from( + "cluster.local", + ) + .expect("valid domain"), + node_id: replica.into(), + role, + client_port: client_port.into(), + } + } + + #[test] + fn kraft_controllers_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" + ); + } + + #[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) + ); + } + + #[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 quorum_bootstrap_servers = kraft_controllers(&pod_descriptors); + + 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..66426b604 --- /dev/null +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -0,0 +1,157 @@ +//! 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. +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", + }) +} + +/// The broker startup and liveness probe. +/// Combines 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`). +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", + }) +} + +/// The controller startup probe. +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", + }) +} + +/// The controller readiness probe. +/// +/// 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", + }) +} + +/// The controller liveness probe. +/// +/// 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 d166d0352..401cffdd1 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -16,11 +16,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::{ @@ -40,13 +39,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}, @@ -54,8 +55,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, @@ -86,7 +87,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"); @@ -125,7 +125,27 @@ fn common_operator_env_vars( env } +/// Environment variables the operator sets that are common to the `kafka` and `quorum-manager` +/// containers. +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 { @@ -154,6 +174,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, @@ -192,7 +215,6 @@ pub fn build_broker_rolegroup_statefulset( role_group_name, ); - let mut cb_kcat_prober = new_container_builder(BrokerContainer::KcatProber.name()); let mut cb_kafka = new_container_builder(BrokerContainer::Kafka.name()); let mut pod_builder = PodBuilder::new(); @@ -205,7 +227,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, ) @@ -224,14 +245,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. @@ -249,6 +264,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![ @@ -260,12 +287,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 @@ -286,44 +308,10 @@ pub fn build_broker_rolegroup_statefulset( .expect("The mount paths are statically defined and there should be no duplicates.") .add_volume_mount(&*STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) .expect("The mount paths are statically defined and there should be no duplicates.") - .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, - ) - .expect("The mount paths are statically defined and there should be no duplicates.") - .add_volume_mount(&*LISTENER_BROKER_VOLUME_NAME, STACKABLE_LISTENER_BROKER_DIR) - .expect("The mount paths are statically defined and there should be no duplicates.") - // 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, @@ -365,7 +353,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( @@ -389,10 +376,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()); @@ -444,19 +427,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 @@ -469,6 +450,30 @@ 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)?; + + let controller_startup_probe = + probes::controller_tcp_probe(kafka_security.client_port(), 5, 5, 60) + .context(BuildProbeSnafu)?; + + 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![ @@ -479,10 +484,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 @@ -497,27 +499,26 @@ pub fn build_controller_rolegroup_statefulset( .add_volume_mount(&*STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) .expect("The mount paths are statically defined and there should be no duplicates.") .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, @@ -550,6 +551,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, @@ -587,7 +594,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() @@ -635,8 +642,6 @@ fn container_ports(kafka_security: &ValidatedKafkaSecurity) -> Vec, +) -> Option { + if kafka_security.has_kerberos_enabled() { + return None; + } + + let mut cb = new_container_builder(&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(), + ]) + // `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") + .add_env_vars(env) + .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) + .expect("The mount paths are statically defined and there should be no duplicates.") + // `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, + ) + .expect("The mount paths are statically defined and there should be no duplicates.") + // `add-controller` reads this controller's own on-disk `meta.properties` from `log.dirs`. + .add_volume_mount(&*LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) + .expect("The mount paths are statically defined and there should be no duplicates."); + + Some(cb.build()) +} + /// Adds the Vector log-aggregation sidecar container, when the Vector agent is enabled. /// /// Whether Vector is enabled, the per-container log config and the (validated) aggregator @@ -767,6 +827,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}; @@ -780,7 +842,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; @@ -789,7 +850,8 @@ mod tests { /// The user-supplied `envOverrides` must be merged in after all operator-set environment /// variables, so that they can override any of them. `CONTAINERDEBUG_LOG_DIRECTORY` is used - /// as the example here because it is set unconditionally by the operator. + /// as the example here because it is set unconditionally by the operator. The controller + /// builder merges the same [`EnvVarSet`] the same way. #[test] fn env_overrides_override_operator_set_env_vars() { let kafka = minimal_kafka( @@ -852,10 +914,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 @@ -879,44 +942,310 @@ mod tests { replicas: 3 "#, ); - let cluster = validated_cluster(&kafka); - 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"); + validated_cluster(&kafka) + } - let stateful_set = build_controller_rolegroup_statefulset( - &KafkaRole::Controller, - &role_group_name, - &cluster, - &validated_rg, - ) - .expect("the StatefulSet builds"); + #[test] + fn statefulsets_use_ordered_ready_pod_management_for_controllers_only() { + let cluster = kraft_mode_cluster(); + let resources = crate::controller::build::build(&cluster).expect("build succeeds"); + + for (name, expected_policy) in [ + ("simple-kafka-controller-default", "OrderedReady"), + ("simple-kafka-broker-default", "Parallel"), + ] { + let sts = resources + .stateful_sets + .iter() + .find(|sts| sts.metadata.name.as_deref() == Some(name)) + .unwrap_or_else(|| panic!("the {name} StatefulSet is built")); + assert_eq!( + sts.spec + .as_ref() + .expect("the StatefulSet has a spec") + .pod_management_policy + .as_deref(), + Some(expected_policy) + ); + } + } - let env = stateful_set - .spec + 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 - .into_iter() - .find(|container| container.name == "kafka") - .expect("the kafka container exists") - .env - .expect("the kafka container has env vars"); + } - let sleep_seconds: Vec<_> = env + #[test] + fn the_sidecar_joins_the_quorum_and_the_kafka_container_leaves_it_on_stop() { + let cluster = kraft_mode_cluster(); + let containers = controller_containers(&cluster); + let sidecar = containers .iter() - .filter(|env_var| env_var.name == "PRE_STOP_CONTROLLER_SLEEP_SECONDS") + .find(|c| c.name == QUORUM_MANAGER_CONTAINER_NAME.to_string()) + .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")); + assert!(sidecar.lifecycle.is_none()); + + let pre_stop_command = controller_kafka_container(&cluster) + .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 serializing `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 + } + + #[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.to_string()) + .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 losing it + // 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:?}" + ); + } + + 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") + } + + /// The startup and liveness probes share the same check - TCP reachability (a genuinely + /// dead/hung process must still be restarted) plus the broker's JMX `BrokerState` metric + /// reporting `RUNNING` (state `3`); see `probes::broker_running_probe`'s doc comment. + #[test] + fn broker_kafka_container_probes_check_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 startup_probe = container + .startup_probe + .clone() + .expect("the broker kafka container must have a startupProbe"); + assert_eq!(startup_probe.timeout_seconds, Some(5)); + assert_eq!(startup_probe.period_seconds, Some(5)); + assert_eq!(startup_probe.failure_threshold, Some(60)); + + 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 startup probe is a plain TCP check, while the liveness probe additionally inspects + /// the local Raft state and fails specifically on `unattached` + #[test] + fn controller_kafka_container_probes_check_tcp_and_raft_state() { + 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 + .clone() + .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!( - sleep_seconds.len(), - 1, - "the override must replace the operator-set value, not duplicate it" + tcp_socket.port, + IntOrString::Int(client_port.clone().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)); + + let liveness_probe = container + .liveness_probe + .clone() + .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)); + + let readiness_probe = container + .readiness_probe + .expect("the controller kafka container must have a readinessProbe"); + let exec = readiness_probe + .exec + .expect("the readinessProbe must be an exec check"); + assert_eq!( + exec.command.expect("exec has a 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!(sleep_seconds[0].value.as_deref(), Some("42")); } } diff --git a/rust/operator-binary/src/controller/build/scripts/controller-quorum-manager-loop.sh b/rust/operator-binary/src/controller/build/scripts/controller-quorum-manager-loop.sh new file mode 100644 index 000000000..915d9aacf --- /dev/null +++ b/rust/operator-binary/src/controller/build/scripts/controller-quorum-manager-loop.sh @@ -0,0 +1,140 @@ +#!/usr/bin/env bash +# The `quorum-manager` sidecar's main loop: while this controller's local Raft state is +# `observer`, admit it into the KRaft voter set. +# +# If the controller is `leader`, `follower` or `unattched` it does nothing. +# +# It has a special case handling when the second controller is added to the voter list. +# In that case, it polls the metrics endpoint `STABILITY_REQUIRES_POLLS` times before +# adding the controller. +# This is a precaution mechanism to ensure that a quorum with two voters stays healthy. +# A quorum with two voters is problematic in Kraft because none of them should fail. +# KRaft redundancy really only starts at a quorum of 3. +# +# Inputs: +# REPLICA_ID this pod's KRaft `node.id` +# BOOTSTRAP_SERVERS the `controller.quorum.bootstrap.servers` to talk to +# QUORUM_CLI path to `kafka-metadata-quorum.sh` +# ADMIN_CLIENT_CONFIG `--command-config` used for read-only `describe` calls +# ADD_CONTROLLER_CONFIG `--command-config` used for `add-controller` (see +# `ADD_CONTROLLER_PROPERTIES_PATH`: it must also carry this +# controller's own `node.id`/listener config) +# METRICS_URL this controller's own Prometheus metrics endpoint +# CLI_TIMEOUT_SECONDS wall-clock bound for a single CLI call +# CLI_KILL_AFTER_SECONDS grace period before `timeout` escalates to `SIGKILL` +# POLL_INTERVAL_SECONDS pause between two iterations +# STABILITY_REQUIRED_POLLS consecutive healthy polls required before the *first* voter +# is joined by a second one (see below) +# VOTER_STALE_FETCH_SECONDS how long an existing voter may go without fetching before the +# quorum counts as degraded +# +# This loop runs forever and never exits non-zero on its own: a deferred admission is always +# retried on the next poll. + +set -uo pipefail + +: "${REPLICA_ID:?must be set by the operator-generated preamble}" +: "${BOOTSTRAP_SERVERS:?must be set by the operator-generated preamble}" +: "${QUORUM_CLI:?must be set by the operator-generated preamble}" +: "${ADMIN_CLIENT_CONFIG:?must be set by the operator-generated preamble}" +: "${ADD_CONTROLLER_CONFIG:?must be set by the operator-generated preamble}" +: "${METRICS_URL:?must be set by the operator-generated preamble}" +: "${CLI_TIMEOUT_SECONDS:?must be set by the operator-generated preamble}" +: "${CLI_KILL_AFTER_SECONDS:?must be set by the operator-generated preamble}" +: "${POLL_INTERVAL_SECONDS:?must be set by the operator-generated preamble}" +: "${STABILITY_REQUIRED_POLLS:?must be set by the operator-generated preamble}" +: "${VOTER_STALE_FETCH_SECONDS:?must be set by the operator-generated preamble}" + +# Logs a single line, prefixed with an RFC 3339 UTC timestamp. +log() { + echo "$(date -u +%Y-%m-%dT%H:%M:%SZ) $*" +} + +ADD_CONTROLLER_PID="" + +handle_term_signal() { + [ -n "$ADD_CONTROLLER_PID" ] && kill -TERM "$ADD_CONTROLLER_PID" 2>/dev/null + exit 0 +} + +trap 'handle_term_signal' TERM + +# This controller's own Raft state, as reported by its metrics endpoint. Empty when the +# endpoint could not be scraped (the Kafka process is still starting, or already gone). +local_raft_state() { + curl -s --max-time 5 --connect-timeout 2 "$METRICS_URL" \ + | grep -oE 'kafka_server_raft_metrics_current_state\{state="[a-z]+",?\}' \ + | grep -oE '"[a-z]+"' \ + | tr -d '"' +} + +# The current voter rows of `describe --replication`, one per line, header dropped. +# +# Observers carry a replica state of their own and are deliberately excluded: this pod is +# itself an observer while it waits to be admitted. +current_voters() { + timeout --kill-after="$CLI_KILL_AFTER_SECONDS" "$CLI_TIMEOUT_SECONDS" "$QUORUM_CLI" \ + --bootstrap-controller "$BOOTSTRAP_SERVERS" --command-config "$ADMIN_CLIENT_CONFIG" \ + describe --replication 2>/dev/null \ + | tail -n +2 \ + | awk '$NF == "Leader" || $NF == "Follower"' +} + +# Node ids of voters that have not fetched recently enough to be considered alive. +# +# `LastFetchTimestamp` (column 5) is epoch milliseconds as observed by the *leader*, compared +# here against this pod's own clock. Nodes in a Kubernetes cluster are expected to be roughly +# time-synchronised, and VOTER_STALE_FETCH_SECONDS is generous enough to absorb the usual +# skew; this is a liveness heuristic, not a correctness mechanism. +stale_voters() { + local voters=$1 now_ms + now_ms=$(date +%s%3N) + + echo "$voters" | awk -v now="$now_ms" -v max_age="$((VOTER_STALE_FETCH_SECONDS * 1000))" \ + '(now - $5) > max_age { print $1 }' +} + +log "Starting KRaft voter admission loop against bootstrap servers: $BOOTSTRAP_SERVERS" + +# Consecutive polls that found this controller up and reporting `observer`. Reset by anything +# that interrupts that run, so a flapping controller never accumulates a stable streak. +stable_polls=0 + +while true; do + state=$(local_raft_state) + + if [ -z "$state" ]; then + log "Could not determine local Raft state (metrics scrape returned nothing), will retry" + stable_polls=0 + elif [ "$state" != "observer" ]; then + log "Local Raft state is '$state', nothing to do" + stable_polls=0 + else + stable_polls=$((stable_polls + 1)) + + voters=$(current_voters) + voter_count=$(echo "$voters" | grep -c .) + + if [ "$voter_count" -eq 0 ]; then + log "Local Raft state is observer, but the quorum could not be described (unreachable, or its output was unrecognized); deferring add-controller" + elif [ -n "$(stale_voters "$voters")" ]; then + log "Local Raft state is observer, but the existing quorum is degraded (voter(s) $(stale_voters "$voters" | tr '\n' ' ')have not fetched within ${VOTER_STALE_FETCH_SECONDS}s); deferring add-controller rather than perturbing it" + elif [ "$voter_count" -eq 1 ] && [ "$stable_polls" -lt "$STABILITY_REQUIRED_POLLS" ]; then + # Joining the single existing voter makes both nodes load-bearing, so this controller + # has to prove it stays up first. Waiting costs nothing: the quorum stays at one voter. + log "Local Raft state is observer and the quorum has a single voter; proving stability before joining it ($stable_polls/$STABILITY_REQUIRED_POLLS consecutive healthy polls)" + else + log "Local Raft state is observer, attempting add-controller..." + timeout --kill-after="$CLI_KILL_AFTER_SECONDS" "$CLI_TIMEOUT_SECONDS" "$QUORUM_CLI" \ + --bootstrap-controller "$BOOTSTRAP_SERVERS" --command-config "$ADD_CONTROLLER_CONFIG" \ + add-controller & + ADD_CONTROLLER_PID=$! + wait "$ADD_CONTROLLER_PID" \ + || log "add-controller attempt failed (this is expected if it already succeeded or a leader election is in progress), will retry" + ADD_CONTROLLER_PID="" + fi + fi + + sleep "$POLL_INTERVAL_SECONDS" & + wait $! +done diff --git a/rust/operator-binary/src/controller/build/scripts/controller-remove-self-pre-stop.sh b/rust/operator-binary/src/controller/build/scripts/controller-remove-self-pre-stop.sh new file mode 100644 index 000000000..39eec692c --- /dev/null +++ b/rust/operator-binary/src/controller/build/scripts/controller-remove-self-pre-stop.sh @@ -0,0 +1,99 @@ +#!/usr/bin/env bash +# The `preStop` hook of a KRaft controller pod's `kafka` container: remove this pod from the +# KRaft voter set before it terminates. +# +# The last pod (the one with the lowest `node.id`) is never removed as that would effectively +# lead to all cluster state being lost. +# +# Inputs: +# REPLICA_ID this pod's KRaft `node.id` +# BOOTSTRAP_SERVERS the `controller.quorum.bootstrap.servers` to talk to +# QUORUM_CLI path to `kafka-metadata-quorum.sh` +# ADMIN_CLIENT_CONFIG `--command-config` file passed to that CLI +# CLI_TIMEOUT_SECONDS wall-clock bound for a single CLI call +# CLI_KILL_AFTER_SECONDS grace period before `timeout` escalates to `SIGKILL` +# REMOVAL_DEADLINE_SECONDS total budget for retrying the removal +# RETRY_INTERVAL_SECONDS pause between two attempts +# +# Always exits 0 (a missing input aside, which is an operator bug): a failed or stuck removal +# must never be the reason a pod fails to terminate. + +set -uo pipefail + +: "${REPLICA_ID:?must be set by the operator-generated preStop preamble}" +: "${BOOTSTRAP_SERVERS:?must be set by the operator-generated preStop preamble}" +: "${QUORUM_CLI:?must be set by the operator-generated preStop preamble}" +: "${ADMIN_CLIENT_CONFIG:?must be set by the operator-generated preStop preamble}" +: "${CLI_TIMEOUT_SECONDS:?must be set by the operator-generated preStop preamble}" +: "${CLI_KILL_AFTER_SECONDS:?must be set by the operator-generated preStop preamble}" +: "${REMOVAL_DEADLINE_SECONDS:?must be set by the operator-generated preStop preamble}" +: "${RETRY_INTERVAL_SECONDS:?must be set by the operator-generated preStop preamble}" + +# Logs a single line, prefixed with an RFC 3339 UTC timestamp. +log() { + echo "$(date -u +%Y-%m-%dT%H:%M:%SZ) $*" +} + +quorum_cli() { + timeout --kill-after="$CLI_KILL_AFTER_SECONDS" "$CLI_TIMEOUT_SECONDS" "$QUORUM_CLI" \ + --bootstrap-controller "$BOOTSTRAP_SERVERS" --command-config "$ADMIN_CLIENT_CONFIG" "$@" +} + +# One removal attempt. +# +# Returns 0 when there is nothing left to do - either this pod was removed, or it never was +# (or no longer is) a voter, or removing it would leave zero voters. Returns 1 when the +# attempt was inconclusive and is worth retrying while the deadline holds. +attempt_removal() { + local describe voters total_voters directory_id + + describe=$(quorum_cli describe --replication 2>/dev/null) + if [ -z "$describe" ]; then + log "Could not describe the quorum (unreachable or the call timed out), will retry" + return 1 + fi + + # Skip the table header, then keep the rows of actual voters. Observers carry a replica + # state of their own and must not be counted here. + voters=$(echo "$describe" | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"') + total_voters=$(echo "$voters" | grep -c .) + if [ "$total_voters" -eq 0 ]; then + log "Could not identify any voters in the describe output (unrecognized format), skipping removal for safety and retrying..." + return 1 + fi + + # Fewer than two voters means this pod is the last one, so removing it would leave zero. + # This can never become safe later during this pod's own termination - nothing else will + # add a voter on its behalf - so give up instead of retrying until the deadline. + if [ "$total_voters" -lt 2 ]; then + log "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)" + return 0 + fi + + directory_id=$(echo "$voters" | awk -v id="$REPLICA_ID" '$1 == id { print $2 }') + if [ -z "$directory_id" ]; then + log "Could not find own node $REPLICA_ID among current voters (already removed?), nothing to do" + return 0 + fi + + log "Removing self (node $REPLICA_ID, directory $directory_id) from the voter set..." + if ! quorum_cli remove-controller \ + --controller-id "$REPLICA_ID" --controller-directory-id "$directory_id"; then + log "remove-controller attempt failed, will retry if time remains" + return 1 + fi + + return 0 +} + +DEADLINE=$((SECONDS + REMOVAL_DEADLINE_SECONDS)) +while [ "$SECONDS" -lt "$DEADLINE" ]; do + if attempt_removal; then + exit 0 + fi + sleep "$RETRY_INTERVAL_SECONDS" +done + +# Loud on purpose (`ERROR:`, so it is greppable and alertable in the container logs) +log "ERROR: could not remove self (node $REPLICA_ID) from the voter set before terminating (every attempt within ${REMOVAL_DEADLINE_SECONDS}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" +exit 0 diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index 865be8a76..b9eb650af 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -51,8 +51,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"; -constant!(STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: VolumeName = "tls-kafka-internal"); +pub(crate) const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &str = "/stackable/tls-kafka-internal"; +constant!(pub(crate) STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: VolumeName = "tls-kafka-internal"); const STACKABLE_TLS_KAFKA_SERVER_DIR: &str = "/stackable/tls-kafka-server"; constant!(STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME: VolumeName = "tls-kafka-server"); // directories @@ -218,6 +218,24 @@ 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`. +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. /// @@ -228,13 +246,13 @@ pub fn client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Opti 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, @@ -242,7 +260,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) .expect("The mount paths are statically defined and there should be no duplicates."); // Keystores fore the kafka container @@ -658,7 +676,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::{ @@ -731,7 +749,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"), @@ -904,6 +922,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..c41d31a86 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -101,6 +101,20 @@ 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, + + #[snafu(display( + "the `spec.controllers` role is required in KRaft mode; brokers have no metadata quorum \ + to join without it" + ))] + MissingKraftControllerRole, } /// Validated logging configuration for a Kafka role group's Kafka and (optional) Vector @@ -261,8 +275,16 @@ pub fn validate( ); role_group_configs.insert(KafkaRole::Broker, broker_groups); - // Controllers are optional: ZooKeeper-mode clusters have none, in which case they are simply - // absent from both maps and not reconciled. + let metadata_manager = kafka + .effective_metadata_manager() + .context(InvalidMetadataManagerSnafu)?; + + // Controllers are optional in ZooKeeper mode only. + // In KRaft mode they are mandatory. + if metadata_manager == crate::crd::MetadataManager::KRaft && kafka.spec.controllers.is_none() { + return MissingKraftControllerRoleSnafu.fail(); + } + if let Some(controller_role) = kafka.spec.controllers.as_ref() { let controller_groups = validate_role_group_configs( controller_role, @@ -273,6 +295,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 +325,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 +448,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 +572,176 @@ mod tests { let env = inject_cluster_id(EnvVarSet::new(), None); assert_eq!(cluster_id_value(&env), None); } + + #[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:?}" + ); + } + + #[test] + fn kraft_mode_rejects_missing_controller_role() { + 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: 4.1.1 + brokers: + roleGroups: + default: + replicas: 1 + "#, + ); + + let result = crate::controller::test_support::validate_err(&kafka); + let Err(error) = result else { + panic!("validate should reject a KRaft cluster without a controllers role"); + }; + + assert!( + matches!(error, Error::MissingKraftControllerRole), + "expected MissingKraftControllerRole, got: {error:?}" + ); + } + + #[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 + #[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 5005122fa..3402151fa 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -334,42 +334,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 7ff41f9b7..6b50930e6 100644 --- a/rust/operator-binary/src/crd/role/broker.rs +++ b/rust/operator-binary/src/crd/role/broker.rs @@ -37,14 +37,12 @@ constant!(DEFAULT_LISTENER_CLASS: ListenerClassName = "cluster-internal"); #[strum(serialize_all = "kebab-case")] pub enum BrokerContainer { Vector, - KcatProber, Kafka, } // Typed container names. They must match the strum `Display` (kebab-case) of the variants above, // which is pinned by a unit test. constant!(VECTOR_CONTAINER_NAME: ContainerName = "vector"); -constant!(KCAT_PROBER_CONTAINER_NAME: ContainerName = "kcat-prober"); constant!(KAFKA_CONTAINER_NAME: ContainerName = "kafka"); impl BrokerContainer { @@ -52,7 +50,6 @@ impl BrokerContainer { pub fn name(&self) -> &'static ContainerName { match self { BrokerContainer::Vector => &VECTOR_CONTAINER_NAME, - BrokerContainer::KcatProber => &KCAT_PROBER_CONTAINER_NAME, BrokerContainer::Kafka => &KAFKA_CONTAINER_NAME, } } @@ -128,7 +125,6 @@ mod tests { // Test that dereferencing the constants does not panic. let _ = *DEFAULT_LISTENER_CLASS; let _ = *VECTOR_CONTAINER_NAME; - let _ = *KCAT_PROBER_CONTAINER_NAME; let _ = *KAFKA_CONTAINER_NAME; } 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-kraft/90-assert.yaml.j2 b/tests/templates/kuttl/smoke-kraft/90-assert.yaml.j2 new file mode 100644 index 000000000..c727dd508 --- /dev/null +++ b/tests/templates/kuttl/smoke-kraft/90-assert.yaml.j2 @@ -0,0 +1,41 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +commands: + - script: kubectl -n $NAMESPACE wait --for=condition=stopped kafkaclusters.kafka.stackable.tech/test-kafka --timeout 601s +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-default +status: + replicas: 0 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-custom-log-config +status: + replicas: 0 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-automatic-log-config +status: + replicas: 0 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-controller-custom-log-config +status: + replicas: 0 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-controller-automatic-log-config +status: + replicas: 0 diff --git a/tests/templates/kuttl/smoke-kraft/90-stop-kafka.yaml.j2 b/tests/templates/kuttl/smoke-kraft/90-stop-kafka.yaml.j2 new file mode 100644 index 000000000..20a0eac68 --- /dev/null +++ b/tests/templates/kuttl/smoke-kraft/90-stop-kafka.yaml.j2 @@ -0,0 +1,13 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +timeout: 600 +--- +apiVersion: kafka.stackable.tech/v1alpha1 +kind: KafkaCluster +metadata: + name: test-kafka +spec: + clusterOperation: + stopped: true + reconciliationPaused: false 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