wuchong commented on code in PR #3400: URL: https://github.com/apache/fluss/pull/3400#discussion_r3382089833
########## fluss-dist/src/main/resources/bin/readiness-check.sh: ########## @@ -0,0 +1,249 @@ +#!/usr/bin/env bash + +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# ============================================================================== +# Fluss Readiness Check Script +# ============================================================================== +# +# Two-step readiness probe for Kubernetes StatefulSet rolling upgrades: +# +# Step 1 — Local TCP port check: verify this TabletServer process is alive +# and has bound its RPC port. Fast, no external dependency. +# +# Step 2 — Cluster health check: query the Coordinator's Cluster Health API +# and pass only if status is GREEN. +# YELLOW/RED/UNKNOWN means recovery is incomplete — block the upgrade. +# +# Both steps must pass for the pod to be marked Ready. +# +# Exit codes (from ClusterHealthReadinessCheck.java): +# 0 = Ready (status GREEN) +# 1 = Not ready (status YELLOW, RED, UNKNOWN, or Coordinator unreachable) +# 2 = API unsupported → shell handles grace period then TCP fallback +# +# Environment variables (set by helm template or container spec): +# FLUSS_HOME - Fluss installation directory +# FLUSS_CONF_DIR - Configuration directory (default: $FLUSS_HOME/conf) +# READINESS_TIMEOUT_MS - Timeout for Health API call (default: 5000) +# READINESS_TCP_HOST - Host for TCP check (default: $POD_IP or 127.0.0.1) +# READINESS_TCP_PORT - Port for TCP check (default: 9124) +# READINESS_BOOTSTRAP_SERVERS - Coordinator bootstrap address +# READINESS_GRACE_SECS - Grace period when API is unsupported (default: 60) +# ============================================================================== + +set -o pipefail + +# ---- Configuration ---- + +FLUSS_HOME="${FLUSS_HOME:-/opt/fluss}" +FLUSS_CONF_DIR="${FLUSS_CONF_DIR:-${FLUSS_HOME}/conf}" +TIMEOUT_MS="${READINESS_TIMEOUT_MS:-5000}" +TCP_HOST="${READINESS_TCP_HOST:-${POD_IP:-127.0.0.1}}" +TCP_PORT="${READINESS_TCP_PORT:-9124}" +BOOTSTRAP_SERVERS="${READINESS_BOOTSTRAP_SERVERS:-}" +GRACE_SECS="${READINESS_GRACE_SECS:-60}" + +# Marker files for tracking state across probe invocations +MARKER_DIR="/tmp/fluss-readiness" +FIRST_READY_MARKER="${MARKER_DIR}/first-ready" +API_UNSUPPORTED_SINCE="${MARKER_DIR}/api-unsupported-since" + +mkdir -p "${MARKER_DIR}" + +# ---- Helper Functions ---- + +# Step 1: TCP port check (local liveness) +check_tcp() { + local host="${TCP_HOST}" + local port="${1:-${TCP_PORT}}" + (echo > /dev/tcp/"${host}"/"${port}") >/dev/null 2>&1 + return $? +} + +# Step 2: Run the Java ClusterHealthReadinessCheck CLI tool (cluster health check) +run_recovery_check() { + local conf_dir="$1" + local timeout_ms="$2" + + # Construct classpath (same logic as config.sh) + local classpath="" + local fluss_server_jar="" + while IFS= read -r -d '' jarfile; do + if [[ "$jarfile" =~ .*/fluss-server[^/]*.jar$ ]]; then + fluss_server_jar="$jarfile" + elif [[ -z "$classpath" ]]; then + classpath="$jarfile" + else + classpath="${classpath}:${jarfile}" + fi + done < <(find "${FLUSS_HOME}/lib" ! -type d -name '*.jar' -print0 | sort -z) + + if [[ -n "$fluss_server_jar" ]]; then + classpath="${classpath}:${fluss_server_jar}" + fi + + if [[ -z "$classpath" ]]; then + echo "[readiness-check] ERROR: No jars found in ${FLUSS_HOME}/lib" + return 3 + fi + + # Find Java + local java_cmd="java" + if [[ -n "${JAVA_HOME}" && -x "${JAVA_HOME}/bin/java" ]]; then + java_cmd="${JAVA_HOME}/bin/java" + fi + + # Build optional --bootstrapServers argument + local bootstrap_arg="" + if [[ -n "${BOOTSTRAP_SERVERS}" ]]; then + bootstrap_arg="--bootstrapServers ${BOOTSTRAP_SERVERS}" + fi + + # Run the check (suppress JVM startup noise, only care about exit code + output) + local output + output=$("${java_cmd}" \ + -XX:+IgnoreUnrecognizedVMOptions \ + -Xmx64m \ + -classpath "${classpath}" \ + org.apache.fluss.dist.ClusterHealthReadinessCheck \ + --configDir "${conf_dir}" \ + --timeoutMs "${timeout_ms}" \ + ${bootstrap_arg} 2>&1) + local exit_code=$? + + # Log output to container's main process stderr (visible in kubectl logs) + if [[ -n "$output" ]]; then + echo "$output" >&2 + if [[ -w /proc/1/fd/2 ]]; then + echo "[readiness-probe] $output" > /proc/1/fd/2 + fi + fi + + return $exit_code +} + +# Record when API unsupported was first detected +mark_api_unsupported() { + if [[ ! -f "${API_UNSUPPORTED_SINCE}" ]]; then + date +%s > "${API_UNSUPPORTED_SINCE}" + fi +} + +# Clear API-unsupported marker +clear_api_unsupported() { + rm -f "${API_UNSUPPORTED_SINCE}" +} + +# Check if grace period for API-unsupported has elapsed +is_grace_period_elapsed() { + if [[ ! -f "${API_UNSUPPORTED_SINCE}" ]]; then + return 1 # not elapsed (no marker) + fi + local since + since=$(cat "${API_UNSUPPORTED_SINCE}") + local now + now=$(date +%s) + local elapsed=$(( now - since )) + if [[ $elapsed -ge ${GRACE_SECS} ]]; then + return 0 # elapsed + fi + return 1 # not elapsed yet +} + +# ---- Main Logic ---- + +# Helper: log to container's main process (visible in kubectl logs) +log_to_main() { + echo "$1" >&2 + if [[ -w /proc/1/fd/2 ]]; then + echo "$1" > /proc/1/fd/2 + fi +} + +# ---- Step 1: Local TCP port check ---- +# If the local TS process hasn't bound its port yet, fail immediately. +# No need to query the Coordinator for cluster state. +if ! check_tcp; then + exit 1 +fi + +# ---- Step 2: Cluster health check ---- + +# First-boot detection: +# On the very first startup (no previous data), there's nothing to recover. +# We use a marker file: once the cluster has been "ready" at least once, we +# know subsequent probes after a restart are "upgrade" scenarios that need +# the full recovery gate. +if [[ ! -f "${FIRST_READY_MARKER}" ]]; then + run_recovery_check "${FLUSS_CONF_DIR}" "${TIMEOUT_MS}" + local_exit=$? + + case $local_exit in + 0) + log_to_main "[readiness-check] First boot: GREEN, marking ready" + touch "${FIRST_READY_MARKER}" + clear_api_unsupported + exit 0 + ;; + 1) + log_to_main "[readiness-check] First boot: not ready (exit 1)" + exit 1 + ;; + 2) + log_to_main "[readiness-check] First boot: API unsupported, TCP-only fallback" + touch "${FIRST_READY_MARKER}" + exit 0 + ;; + *) + log_to_main "[readiness-check] Config error, falling back to TCP-only" + exit 0 + ;; + esac +fi + +# Upgrade/restart scenario (marker file exists = has been ready before) +run_recovery_check "${FLUSS_CONF_DIR}" "${TIMEOUT_MS}" +local_exit=$? + +case $local_exit in + 0) + log_to_main "[readiness-check] Upgrade check: PASSED (exit 0)" + clear_api_unsupported + exit 0 + ;; + 1) + log_to_main "[readiness-check] Upgrade check: BLOCKED (exit 1) — waiting for recovery" + clear_api_unsupported + exit 1 + ;; + 2) + mark_api_unsupported + if is_grace_period_elapsed; then + log_to_main "[readiness-check] API unsupported for >${GRACE_SECS}s, TCP fallback" + exit 0 + fi + log_to_main "[readiness-check] API unsupported, waiting (grace period)" + exit 1 + ;; + *) + log_to_main "[readiness-check] Config error (exit $local_exit), treating as not ready" + exit 1 + ;; +esac Review Comment: Suggest removing the `READINESS_GRACE_SECS` grace-period logic entirely (this `mark_api_unsupported` / `is_grace_period_elapsed` machinery and the `--default 60` plumbing). The grace period is meant to handle the mixed-version window during a rolling upgrade — "wait a bit, maybe the Coordinator finishes upgrading and starts supporting the API." But it doesn't actually help, and it has a real cost: - On the **first upgrade from a version that predates this API**, the whole cluster is on the old Coordinator, so the API returns `exit 2` (unsupported) for the *entire* upgrade. Every tablet-server pod is then forced to sit at NotReady for the full `GRACE_SECS` (default 60s) before falling back to TCP-only. Since the StatefulSet rolls serially, that's `60s × pod count` of pure dead time added to the upgrade, with zero protection gained (the API genuinely cannot be available in this window). - The case the grace period is *trying* to protect — "Coordinator temporarily unreachable / still electing" — should already be `exit 1` (not ready, retry next cycle), not `exit 2`. A clean `UnsupportedVersionException` means the server is up and definitively does not support the API, so there is nothing to wait for: we should TCP-fallback immediately. So `exit 2` can just latch the fast path right away, same as the config-error `*` branch below it. This removes the grace-period env var, the two marker helpers, and the first-upgrade slowdown. ########## helm/templates/sts-tablet.yaml: ########## @@ -128,12 +130,21 @@ spec: tcpSocket: port: {{ .Values.listeners.client.port }} readinessProbe: - failureThreshold: 100 - timeoutSeconds: 1 - initialDelaySeconds: 10 - periodSeconds: 3 - tcpSocket: - port: {{ .Values.listeners.client.port }} + failureThreshold: {{ .Values.tablet.readinessProbe.failureThreshold | default 200 }} + timeoutSeconds: {{ .Values.tablet.readinessProbe.timeoutSeconds | default 10 }} + initialDelaySeconds: {{ .Values.tablet.readinessProbe.initialDelaySeconds | default 15 }} + periodSeconds: {{ .Values.tablet.readinessProbe.periodSeconds | default 5 }} + exec: + command: + - /bin/bash + - -c + - | + export FLUSS_SERVER_ID=${POD_NAME##*-} + export READINESS_TIMEOUT_MS={{ .Values.tablet.readinessProbe.rpcTimeoutMs | default 5000 }} + export READINESS_TCP_PORT={{ .Values.listeners.client.port }} + export READINESS_BOOTSTRAP_SERVERS="coordinator-server-hs.${POD_NAMESPACE}.svc.cluster.local:{{ .Values.listeners.client.port }}" Review Comment: Will the coordinator address change in scenarios such as upgrades, restarts, or failover migrations? If it does, could this prevent the tablet server from connecting to the coordinator during its next restart or upgrade? ########## fluss-rpc/src/main/proto/FlussApi.proto: ########## @@ -1369,4 +1369,14 @@ message PbLiteralValue { optional int64 timestamp_millis_value = 11; // Epoch millis optional int32 timestamp_nano_of_millis_value = 12; // Nano of millis optional bytes decimal_bytes = 13; // Serialized decimal (non-compact mode) +} + +message GetClusterHealthRequest { } + +message GetClusterHealthResponse { + required int32 num_replicas = 1; + required int32 in_sync_replicas = 2; + required int32 num_leader_replicas = 3; + required int32 active_leader_replicas = 4; + required int32 status = 5; // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3 Review Comment: nit: move the RPC messages to before the line of ``` // --------------- Inner classes ---------------- ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
