From 0d410a4b26e51f3d4fd0ccc375bae9773217696c Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:03:22 +0200 Subject: [PATCH 1/8] fix: readiness probe checks cluster CONNECTED status in response body NiFi's /health/cluster management-server endpoint returns HTTP 200 for both CONNECTING and CONNECTED nodes (only DISCONNECTED etc. get a non-2xx status), so a bare `curl --fail` could not tell a node still joining the cluster from one that has actually joined it. Verified against a live NiFi 2.9.0 node in minikube: /health/cluster returned 200 with body "Cluster Status: CONNECTING" while joining, and 200 with "Cluster Status: CONNECTED" once fully joined. The readiness probe now greps the response body for "Cluster Status: CONNECTED" instead of only checking the return code. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 3 +- .../src/controller/build/resource/probes.rs | 46 ++++++++++++++++++- 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 377dd07a..a7f6082a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ All notable changes to this project will be documented in this file. `secrets` ([#974]). - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#975]). - NiFi startup and readiness probes now use the local management server's `/health` and - `/health/cluster` endpoints instead of a bare TCP check ([#976]). + `/health/cluster` endpoints instead of a bare TCP check ([#976], [#981]). ### Fixed @@ -36,6 +36,7 @@ All notable changes to this project will be documented in this file. [#974]: https://github.com/stackabletech/nifi-operator/pull/974 [#975]: https://github.com/stackabletech/nifi-operator/pull/975 [#976]: https://github.com/stackabletech/nifi-operator/pull/976 +[#981]: https://github.com/stackabletech/nifi-operator/pull/981 ## [26.7.0] - 2026-07-21 diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index 19327764..a44c65fe 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -28,6 +28,24 @@ fn management_health_exec(path: &str) -> ExecAction { ]), } } + +/// NiFi's `/health/cluster` endpoint returns HTTP 200 for both `CONNECTING` +/// and `CONNECTED`, so the readiness probe greps for that instead of only +/// checking the return code. +fn management_cluster_connected_exec() -> ExecAction { + ExecAction { + command: Some(vec![ + "/bin/bash".to_string(), + "-euo".to_string(), + "pipefail".to_string(), + "-c".to_string(), + format!( + "curl --fail --silent --show-error http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}/health/cluster | grep -q 'Cluster Status: CONNECTED'" + ), + ]), + } +} + pub fn management_startup_probe() -> Probe { Probe { initial_delay_seconds: Some(10), @@ -43,7 +61,7 @@ pub fn management_readiness_probe() -> Probe { period_seconds: Some(10), timeout_seconds: Some(5), failure_threshold: Some(3), - exec: Some(management_health_exec("/health/cluster")), + exec: Some(management_cluster_connected_exec()), ..Probe::default() } } @@ -113,6 +131,32 @@ mod tests { ); } + #[test] + fn readiness_probe_greps_body_for_connected_status() { + // NiFi's /health/cluster endpoint returns HTTP 200 for both + // CONNECTING and CONNECTED nodes, so the return code alone can't + // distinguish a node that is still joining from one that has + // actually joined. The probe must inspect the response body. + let probe = management_readiness_probe(); + + let command = probe + .exec + .expect("readiness probe must be an exec probe") + .command + .expect("exec action must have a command"); + let script = command.last().expect("bash -c script argument"); + + assert!( + script.contains("grep") && script.contains("Cluster Status: CONNECTED"), + "expected the probe to grep the response body for \"Cluster Status: CONNECTED\", \ + got: {script}" + ); + assert!( + !script.contains("--output /dev/null"), + "the response body must not be discarded, the probe needs to inspect it: {script}" + ); + } + #[test] fn probes_use_bash_pipefail_wrapper_not_bare_curl_argv() { for probe in [management_startup_probe(), management_readiness_probe()] { From 922b7779106b6a274ccac78716610bc149c16873 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:11:51 +0200 Subject: [PATCH 2/8] refactor: use stackable_operator ProbeBuilder for probe construction Replaces manual Probe struct literals with ProbeBuilder, which validates that duration fields fit into K8s's i32 seconds fields instead of silently truncating. Co-Authored-By: Claude Sonnet 5 --- .../src/controller/build/resource/probes.rs | 103 +++++++++--------- 1 file changed, 51 insertions(+), 52 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index a44c65fe..bdbac1cb 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -6,76 +6,74 @@ //! probes using `curl` from inside the container - a `httpGet` probe cannot //! reach a loopback-only address. -use stackable_operator::k8s_openapi::{ - api::core::v1::{ExecAction, Probe, TCPSocketAction}, - apimachinery::pkg::util::intstr::IntOrString, +use stackable_operator::{ + builder::pod::probe::ProbeBuilder, + k8s_openapi::{ + api::core::v1::{Probe, TCPSocketAction}, + apimachinery::pkg::util::intstr::IntOrString, + }, + shared::time::Duration, }; use crate::controller::build::{ HTTPS_PORT_NAME, MANAGEMENT_SERVER_ADDRESS, MANAGEMENT_SERVER_PORT, }; -fn management_health_exec(path: &str) -> ExecAction { - ExecAction { - command: Some(vec![ - "/bin/bash".to_string(), - "-euo".to_string(), - "pipefail".to_string(), - "-c".to_string(), - format!( - "curl --fail --silent --show-error --output /dev/null http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}{path}" - ), - ]), - } +fn management_health_exec_command(path: &str) -> Vec { + vec![ + "/bin/bash".to_string(), + "-euo".to_string(), + "pipefail".to_string(), + "-c".to_string(), + format!( + "curl --fail --silent --show-error --output /dev/null http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}{path}" + ), + ] } /// NiFi's `/health/cluster` endpoint returns HTTP 200 for both `CONNECTING` /// and `CONNECTED`, so the readiness probe greps for that instead of only /// checking the return code. -fn management_cluster_connected_exec() -> ExecAction { - ExecAction { - command: Some(vec![ - "/bin/bash".to_string(), - "-euo".to_string(), - "pipefail".to_string(), - "-c".to_string(), - format!( - "curl --fail --silent --show-error http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}/health/cluster | grep -q 'Cluster Status: CONNECTED'" - ), - ]), - } +fn management_cluster_connected_exec_command() -> Vec { + vec![ + "/bin/bash".to_string(), + "-euo".to_string(), + "pipefail".to_string(), + "-c".to_string(), + format!( + "curl --fail --silent --show-error http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}/health/cluster | grep -q 'Cluster Status: CONNECTED'" + ), + ] } pub fn management_startup_probe() -> Probe { - Probe { - initial_delay_seconds: Some(10), - period_seconds: Some(10), - timeout_seconds: Some(5), - failure_threshold: Some(20 * 6), - exec: Some(management_health_exec("/health")), - ..Probe::default() - } + ProbeBuilder::exec_command(management_health_exec_command("/health")) + .with_period(Duration::from_secs(10)) + .with_initial_delay(Duration::from_secs(10)) + .with_timeout(Duration::from_secs(5)) + .with_failure_threshold(20 * 6) + .build() + .expect("the startup probe's durations must fit into an i32") } + pub fn management_readiness_probe() -> Probe { - Probe { - period_seconds: Some(10), - timeout_seconds: Some(5), - failure_threshold: Some(3), - exec: Some(management_cluster_connected_exec()), - ..Probe::default() - } + ProbeBuilder::exec_command(management_cluster_connected_exec_command()) + .with_period(Duration::from_secs(10)) + .with_timeout(Duration::from_secs(5)) + .with_failure_threshold(3) + .build() + .expect("the readiness probe's durations must fit into an i32") } pub fn tcp_liveliness_probe() -> Probe { - Probe { - initial_delay_seconds: Some(10), - period_seconds: Some(10), - tcp_socket: Some(TCPSocketAction { - port: IntOrString::String(HTTPS_PORT_NAME.to_string()), - ..TCPSocketAction::default() - }), - ..Probe::default() - } + ProbeBuilder::tcp_socket(TCPSocketAction { + port: IntOrString::String(HTTPS_PORT_NAME.to_string()), + ..Default::default() + }) + .with_period(Duration::from_secs(10)) + .with_initial_delay(Duration::from_secs(10)) + .build() + .expect("the liveliness probe's durations must fit into an i32") } #[cfg(test)] @@ -125,7 +123,8 @@ mod tests { assert_eq!(probe.failure_threshold, Some(3)); assert_eq!(probe.timeout_seconds, Some(5)); assert_eq!( - probe.initial_delay_seconds, None, + probe.initial_delay_seconds, + Some(0), "readiness probe delay is redundant: k8s already suppresses readiness checks \ until the startup probe succeeds" ); From 5c281699525ffc5837c7053c1b540fbde13c5c3f Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:23:15 +0200 Subject: [PATCH 3/8] rename liveliness to liveness --- .../src/controller/build/resource/probes.rs | 8 ++++---- .../src/controller/build/resource/statefulset.rs | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index bdbac1cb..3d3a274c 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -60,12 +60,12 @@ pub fn management_readiness_probe() -> Probe { ProbeBuilder::exec_command(management_cluster_connected_exec_command()) .with_period(Duration::from_secs(10)) .with_timeout(Duration::from_secs(5)) - .with_failure_threshold(3) + .with_failure_threshold(30) .build() .expect("the readiness probe's durations must fit into an i32") } -pub fn tcp_liveliness_probe() -> Probe { +pub fn tcp_liveness_probe() -> Probe { ProbeBuilder::tcp_socket(TCPSocketAction { port: IntOrString::String(HTTPS_PORT_NAME.to_string()), ..Default::default() @@ -73,7 +73,7 @@ pub fn tcp_liveliness_probe() -> Probe { .with_period(Duration::from_secs(10)) .with_initial_delay(Duration::from_secs(10)) .build() - .expect("the liveliness probe's durations must fit into an i32") + .expect("the liveness probe's durations must fit into an i32") } #[cfg(test)] @@ -120,7 +120,7 @@ mod tests { script.contains(&cluster_health_url), "expected curl against /health/cluster, got: {script}" ); - assert_eq!(probe.failure_threshold, Some(3)); + assert_eq!(probe.failure_threshold, Some(30)); assert_eq!(probe.timeout_seconds, Some(5)); assert_eq!( probe.initial_delay_seconds, diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 446f2128..656bb0af 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -61,7 +61,7 @@ use crate::{ group_listener_name, }, probes::{ - management_readiness_probe, management_startup_probe, tcp_liveliness_probe, + management_readiness_probe, management_startup_probe, tcp_liveness_probe, }, }, }, @@ -442,7 +442,7 @@ pub(crate) fn build_node_rolegroup_statefulset( .add_container_port(HTTPS_PORT_NAME, HTTPS_PORT.into()) .add_container_port(PROTOCOL_PORT_NAME, PROTOCOL_PORT.into()) .add_container_port(BALANCE_PORT_NAME, BALANCE_PORT.into()) - .liveness_probe(tcp_liveliness_probe()) + .liveness_probe(tcp_liveness_probe()) .startup_probe(management_startup_probe()) .readiness_probe(management_readiness_probe()) .resources(merged_config.resources.clone().into()); From 5c013e430d7cf742997052d6d81810431f275e65 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:38:17 +0200 Subject: [PATCH 4/8] Update changelog --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a7f6082a..e77e89f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ All notable changes to this project will be documented in this file. `secrets` ([#974]). - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#975]). - NiFi startup and readiness probes now use the local management server's `/health` and - `/health/cluster` endpoints instead of a bare TCP check ([#976], [#981]). + `/health/cluster` endpoints instead of a bare TCP check ([#976], [#982]). ### Fixed @@ -36,7 +36,7 @@ All notable changes to this project will be documented in this file. [#974]: https://github.com/stackabletech/nifi-operator/pull/974 [#975]: https://github.com/stackabletech/nifi-operator/pull/975 [#976]: https://github.com/stackabletech/nifi-operator/pull/976 -[#981]: https://github.com/stackabletech/nifi-operator/pull/981 +[#982]: https://github.com/stackabletech/nifi-operator/pull/982 ## [26.7.0] - 2026-07-21 From 9081c054413cce55b9dd0c003f9b455ef89db919 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:31:47 +0200 Subject: [PATCH 5/8] Apply suggestions from code review Co-authored-by: Sebastian Bernauer --- .../src/controller/build/resource/probes.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index 3d3a274c..bb365f47 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -19,8 +19,8 @@ use crate::controller::build::{ HTTPS_PORT_NAME, MANAGEMENT_SERVER_ADDRESS, MANAGEMENT_SERVER_PORT, }; -fn management_health_exec_command(path: &str) -> Vec { - vec![ +fn management_health_exec_command(path: &str) -> [String; 5] { + [ "/bin/bash".to_string(), "-euo".to_string(), "pipefail".to_string(), @@ -34,8 +34,8 @@ fn management_health_exec_command(path: &str) -> Vec { /// NiFi's `/health/cluster` endpoint returns HTTP 200 for both `CONNECTING` /// and `CONNECTED`, so the readiness probe greps for that instead of only /// checking the return code. -fn management_cluster_connected_exec_command() -> Vec { - vec![ +fn management_cluster_connected_exec_command() -> [String; 5] { + [ "/bin/bash".to_string(), "-euo".to_string(), "pipefail".to_string(), @@ -51,7 +51,8 @@ pub fn management_startup_probe() -> Probe { .with_period(Duration::from_secs(10)) .with_initial_delay(Duration::from_secs(10)) .with_timeout(Duration::from_secs(5)) - .with_failure_threshold(20 * 6) + .with_failure_threshold_duration(Duration::from_minutes_unchecked(20)) + .expect("static period is non-zero") .build() .expect("the startup probe's durations must fit into an i32") } @@ -60,7 +61,8 @@ pub fn management_readiness_probe() -> Probe { ProbeBuilder::exec_command(management_cluster_connected_exec_command()) .with_period(Duration::from_secs(10)) .with_timeout(Duration::from_secs(5)) - .with_failure_threshold(30) + .with_failure_threshold_duration(Duration::from_minutes_unchecked(5)) + .expect("static period is non-zero") .build() .expect("the readiness probe's durations must fit into an i32") } From 8c0dea3a5ab8c4556ce47e876dfb64054b26f6ab Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:51:49 +0200 Subject: [PATCH 6/8] Remove `path` argument --- .../operator-binary/src/controller/build/resource/probes.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index bb365f47..9569f21b 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -19,14 +19,14 @@ use crate::controller::build::{ HTTPS_PORT_NAME, MANAGEMENT_SERVER_ADDRESS, MANAGEMENT_SERVER_PORT, }; -fn management_health_exec_command(path: &str) -> [String; 5] { +fn management_health_exec_command() -> [String; 5] { [ "/bin/bash".to_string(), "-euo".to_string(), "pipefail".to_string(), "-c".to_string(), format!( - "curl --fail --silent --show-error --output /dev/null http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}{path}" + "curl --fail --silent --show-error --output /dev/null http://{MANAGEMENT_SERVER_ADDRESS}:{MANAGEMENT_SERVER_PORT}/health" ), ] } @@ -47,7 +47,7 @@ fn management_cluster_connected_exec_command() -> [String; 5] { } pub fn management_startup_probe() -> Probe { - ProbeBuilder::exec_command(management_health_exec_command("/health")) + ProbeBuilder::exec_command(management_health_exec_command()) .with_period(Duration::from_secs(10)) .with_initial_delay(Duration::from_secs(10)) .with_timeout(Duration::from_secs(5)) From db17391aaf3f68c86cf49dfb3e7b042456939fa7 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 18 Aug 2026 10:31:10 +0200 Subject: [PATCH 7/8] Add comments to explain probe timeouts. --- .../src/controller/build/resource/probes.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index 9569f21b..3813d4a9 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -46,6 +46,19 @@ fn management_cluster_connected_exec_command() -> [String; 5] { ] } +/// The startup probe fails after roughly 20 minutes. +/// +/// Nifi might take a very long time to start up due to the following factors: +/// - JVM cold starts are usually slow. +/// - It expands NAR bundles (each processor/controller-service bundle) into +/// the working directory and builds a classloader per NAR. +/// - It replays/rolls back the FlowFile repository to reconstruct in-flight +/// FlowFile state. +/// If the previous shutdown wasn't clean, or there's a large backlog of +/// in-flight FlowFiles, this replay can take a while. +/// Content and provenance repositories also do startup housekeeping. +/// - Large flow definitions take longer to deserialize and instantiate into +/// the running flow controller graph. pub fn management_startup_probe() -> Probe { ProbeBuilder::exec_command(management_health_exec_command()) .with_period(Duration::from_secs(10)) @@ -57,6 +70,11 @@ pub fn management_startup_probe() -> Probe { .expect("the startup probe's durations must fit into an i32") } +/// The readiness probe fails after roughly 5 minutes. +/// +/// In clustered mode, a node connecting has to talk to the cluster coordinator, +/// participate in flow election/inheritance, and reconcile its local flow against +/// the cluster's. pub fn management_readiness_probe() -> Probe { ProbeBuilder::exec_command(management_cluster_connected_exec_command()) .with_period(Duration::from_secs(10)) @@ -73,7 +91,6 @@ pub fn tcp_liveness_probe() -> Probe { ..Default::default() }) .with_period(Duration::from_secs(10)) - .with_initial_delay(Duration::from_secs(10)) .build() .expect("the liveness probe's durations must fit into an i32") } From 538d1cb9b19efeeddbd68599dfe0e485e576d55f Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 18 Aug 2026 11:23:23 +0200 Subject: [PATCH 8/8] Apply suggestions from code review Co-authored-by: Sebastian Bernauer --- rust/operator-binary/src/controller/build/resource/probes.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index 3813d4a9..bd81638d 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -46,7 +46,7 @@ fn management_cluster_connected_exec_command() -> [String; 5] { ] } -/// The startup probe fails after roughly 20 minutes. +/// We give up on the startup probe after ~20 minutes. /// /// Nifi might take a very long time to start up due to the following factors: /// - JVM cold starts are usually slow. @@ -70,7 +70,7 @@ pub fn management_startup_probe() -> Probe { .expect("the startup probe's durations must fit into an i32") } -/// The readiness probe fails after roughly 5 minutes. +/// We give up on the readiness probe after ~5 minutes. /// /// In clustered mode, a node connecting has to talk to the cluster coordinator, /// participate in flow election/inheritance, and reconcile its local flow against