id: haproxy-backend-drain-gate
namespace: company.team
description: |
Take one server out of an HAProxy backend for maintenance without cutting live connections,
and put it back once its health check passes. A drain only goes ahead when the other servers
can carry the traffic and on-call approves it. The flow sets the server to DRAIN, waits until
its sessions reach zero, then sets MAINT. The enable mode sets READY and waits for UP.
concurrency:
limit: 1
behavior: QUEUE
triggers:
- id: haproxy_change
type: io.kestra.plugin.core.trigger.Webhook
key: "{{ secret('HAPROXY_DRAIN_WEBHOOK_KEY') }}"
description: >-
Start a drain or an enable from a maintenance script or chat bot. Query
parameters override the inputs: `curl -X POST
"http://localhost:8080/api/v1/main/executions/webhook/company.team/haproxy-backend-drain-gate/$KEY?action=enable&backend=web&server=web1"`.
Only action, backend and server can be overridden this way.
- id: nightly_reenable
type: io.kestra.plugin.core.trigger.Schedule
cron: "30 3 * * *"
disabled: true
inputs:
action: enable
description: >-
Nightly backstop: puts the default server back in rotation if a drain was
left behind, for example after a drain timed out and nobody came back to
it. It does nothing when the server is already UP. Shipped disabled,
because a server that is in maintenance on purpose must not be re-enabled
by a timer.
inputs:
- id: action
type: SELECT
values:
- drain
- enable
defaults: drain
description: drain takes the server out of rotation for maintenance after
approval. enable puts it back and waits for its health check to report UP.
- id: backend
type: STRING
defaults: web
description: Name of the HAProxy backend, as written in haproxy.cfg. Letters,
digits, dot, dash and underscore only.
- id: server
type: STRING
defaults: web1
description: Name of the server inside the backend. Letters, digits, dot, dash
and underscore only. A name that does not exist is refused.
- id: min_healthy
type: INT
defaults: 2
description: Servers other than this one that must be UP after the drain. A
drain that would leave fewer is refused. Raise it to protect capacity; set
it to 0 only for a single-server backend.
- id: drain_timeout_seconds
type: INT
defaults: 600
description: How long to wait for the server's sessions to close. Long-lived
clients such as WebSockets or downloads may need more. On timeout the
server stays in DRAIN and the run fails.
- id: poll_seconds
type: INT
defaults: 10
description: How often the flow reads the server's session count and health
while it waits.
- id: health_timeout_seconds
type: INT
defaults: 120
description: How long an enable waits for the health check to report UP. The
check needs a few passes (rise count times the interval), so keep this
above that.
- id: haproxy_host
type: STRING
defaults: 127.0.0.1
description: Host of the HAProxy runtime socket. Bind the socket to loopback or
a management network only, because it has no authentication.
- id: haproxy_port
type: INT
defaults: 9990
description: TCP port of the HAProxy runtime socket, set with `stats socket
ipv4@<addr>:<port> level admin` in the global section.
- id: kestra_url
type: STRING
defaults: http://localhost:8080
description: Base URL of your Kestra UI, used for the approval link in Slack.
variables:
# Webhook calls can override these three from the query string (trigger.parameters holds each query value as a list); manual runs and the schedule use the inputs.
action: "{{ trigger.parameters.action[0] ?? inputs.action }}"
backend: "{{ trigger.parameters.backend[0] ?? inputs.backend }}"
server: "{{ trigger.parameters.server[0] ?? inputs.server }}"
# One read of `show stat` for one backend. Prints the server list and the target's status, sessions and
# the number of other UP servers as a ::{"outputs":...}:: line. HAProxy's runtime API is plain text over TCP.
read_stats: |
set -eu
{ exec 3<>/dev/tcp/$HA_HOST/$HA_PORT; } 2>/dev/null || { echo "HAProxy runtime socket not reachable at $HA_HOST:$HA_PORT" >&2; exit 1; }
printf 'show stat\n' >&3
now=$(date +%s)
start=${HA_START:-$now}
timed_out=false
if [ "${HA_TIMEOUT:-0}" -gt 0 ] && [ $((now - start)) -ge "$HA_TIMEOUT" ]; then timed_out=true; fi
cat <&3 | awk -F, -v be="$HA_BACKEND" -v sv="$HA_SERVER" -v now="$now" -v timed_out="$timed_out" '
NR == 1 { sub(/^# /, ""); for (i = 1; i <= NF; i++) col[$i] = i; next }
$1 == be && $2 != "BACKEND" {
st = $(col["status"]); sc = $(col["scur"]) + 0
list = list (n++ ? "," : "") sprintf("{\"server\":\"%s\",\"status\":\"%s\",\"sessions\":%d}", $2, st, sc)
if ($2 == sv) { tstatus = st; tsess = sc; found = "true" }
else if (st == "UP") up++
}
END {
printf "::{\"outputs\":{\"servers\":[%s],\"found\":%s,\"target_status\":\"%s\",\"target_sessions\":%d,\"others_up\":%d,\"timed_out\":%s,\"now\":%d}}::\n", list, (found == "true" ? "true" : "false"), tstatus, tsess, up, timed_out, now
}'
# Sends one `set server <backend>/<server> state <HA_STATE>` and fails on any reply, because HAProxy answers
# errors (for example "No such server.") with text and success with nothing.
set_state: |
set -eu
{ exec 3<>/dev/tcp/$HA_HOST/$HA_PORT; } 2>/dev/null || { echo "HAProxy runtime socket not reachable at $HA_HOST:$HA_PORT" >&2; exit 1; }
printf 'set server %s/%s state %s\n' "$HA_BACKEND" "$HA_SERVER" "$HA_STATE" >&3
reply=""
read -r -t 5 reply <&3 || true
if [ -n "$reply" ]; then echo "HAProxy refused 'set server $HA_BACKEND/$HA_SERVER state $HA_STATE': $reply" >&2; exit 1; fi
printf '::{"outputs":{"started_at":%s}}::\n' "$(date +%s)"
echo "$HA_BACKEND/$HA_SERVER set to $HA_STATE"
tasks:
- id: validate_input
type: io.kestra.plugin.core.flow.If
description: Names and modes are checked before the socket is touched. A bad
name never reaches HAProxy.
condition: >-
{{ (render(vars.backend) | toJson | jq('if test("^[A-Za-z0-9_.-]+$") then
"ok" else "bad" end')) contains 'bad'
or (render(vars.server) | toJson | jq('if test("^[A-Za-z0-9_.-]+$") then "ok" else "bad" end')) contains 'bad'
or (not (['drain', 'enable'] contains render(vars.action)))
or inputs.min_healthy < 0
or inputs.drain_timeout_seconds < 1
or inputs.poll_seconds < 1
or inputs.health_timeout_seconds < 1 }}
then:
- id: reject_input
type: io.kestra.plugin.core.execution.Fail
errorMessage: >-
Rejected before any change. backend and server may only contain
letters, digits, dot, dash and underscore (got backend '{{
render(vars.backend) }}', server '{{ render(vars.server) }}'). action
must be drain or enable (got '{{ render(vars.action) }}'). min_healthy
and the timeouts must be positive or zero.
- id: read_pool
type: io.kestra.plugin.scripts.shell.Commands
description: Reads every server of the backend from HAProxy (`show stat`).
Retried, because the runtime socket can be restarting.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: ""
HA_TIMEOUT: "0"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: target_check
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.read_pool.vars.found == false }}"
then:
- id: unknown_server
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Server '{{ render(vars.server) }}' is not in backend '{{
render(vars.backend) }}'. Servers HAProxy reports there: {{
outputs.read_pool.vars.servers | jq('[.[].server] | join(\", \")') |
first }}. Check the name, then run again."
- id: if_enable
type: io.kestra.plugin.core.flow.If
description: Enable mode. Puts the server back in rotation and waits for its
health check.
condition: "{{ render(vars.action) == 'enable' }}"
then:
- id: set_ready
type: io.kestra.plugin.scripts.shell.Commands
description: Sets the server to READY. Health checks then decide when it takes
traffic.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_STATE: ready
commands:
- "{{ vars.set_state }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: wait_health
type: io.kestra.plugin.core.flow.LoopUntil
description: Reads the server every poll_seconds until HAProxy reports UP, or
health_timeout_seconds passes.
condition: "{{ outputs.poll_health.vars.target_status == 'UP' or
outputs.poll_health.vars.timed_out == true }}"
checkFrequency:
interval: "PT{{ inputs.poll_seconds }}S"
maxIterations: 1000
failOnMaxReached: false
tasks:
- id: poll_health
type: io.kestra.plugin.scripts.shell.Commands
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: "{{ outputs.set_ready.vars.started_at }}"
HA_TIMEOUT: "{{ inputs.health_timeout_seconds }}"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: confirm_health
type: io.kestra.plugin.scripts.shell.Commands
description: One more read after the wait, so the report and the verdict use the
current state.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: ""
HA_TIMEOUT: "0"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: health_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.confirm_health.vars.target_status != 'UP' }}"
then:
- id: not_up
type: io.kestra.plugin.core.execution.Fail
errorMessage: "{{ render(vars.backend) }}/{{ render(vars.server) }} is READY but
its health check has not reported UP within {{
inputs.health_timeout_seconds }} s (HAProxy status: {{
outputs.confirm_health.vars.target_status }}). It takes no traffic
until the check passes. Check the server's health endpoint, then
look at the HAProxy log."
- id: notify_enabled
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
{{ render(vars.backend) }}/{{ render(vars.server) }} is back in
rotation: HAProxy reports UP with {{
outputs.confirm_health.vars.target_sessions }} session(s). {{
outputs.read_pool.vars.others_up }} other server(s) were UP when the
run started.
- id: if_drain
type: io.kestra.plugin.core.flow.If
description: Drain mode. Checks capacity, asks for approval, drains, waits for
sessions to close, then sets MAINT.
condition: "{{ render(vars.action) == 'drain' }}"
then:
- id: precheck
type: io.kestra.plugin.core.flow.If
description: Refuses before asking anyone when the other servers cannot carry
the traffic.
condition: "{{ outputs.read_pool.vars.others_up < inputs.min_healthy }}"
then:
- id: notify_refused
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
Drain refused for {{ render(vars.backend) }}/{{
render(vars.server) }}: nothing was changed. Only {{
outputs.read_pool.vars.others_up }} other server(s) are UP and the
minimum is {{ inputs.min_healthy }}. Bring a server back, or lower
min_healthy if the capacity loss is acceptable, then run again.
- id: refuse_drain
type: io.kestra.plugin.core.execution.Exit
state: FAILED
- id: ask_approval
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Asks on-call to approve. The approval happens in the Kestra UI,
which shows the capacity numbers.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
Approval needed: drain {{ render(vars.backend) }}/{{
render(vars.server) }} for maintenance. {{
outputs.read_pool.vars.others_up }} other server(s) are UP, the
minimum is {{ inputs.min_healthy }}, and {{
outputs.read_pool.vars.target_sessions }} session(s) are open on it
now. Open {{ inputs.kestra_url }}/ui/main/executions/{{ flow.namespace
}}/{{ flow.id }}/{{ execution.id }} and choose drain. No answer within
an hour changes nothing.
- id: wait_for_decision
type: io.kestra.plugin.core.flow.Pause
description: No answer within an hour means nothing is changed.
pauseDuration: PT1H
behavior: RESUME
onResume:
- id: decision
type: SELECT
values:
- drain
- decline
defaults: decline
description: drain goes ahead. decline leaves the server in rotation, and so
does no answer before the timeout.
- id: decision_gate
type: io.kestra.plugin.core.flow.If
description: Anything but drain counts as a decline, including no answer. A
timed-out Pause leaves the answer empty.
condition: "{{ (outputs.wait_for_decision.onResume.decision ?? '') != 'drain' }}"
then:
- id: notify_declined
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Drain of {{ render(vars.backend) }}/{{ render(vars.server) }} was
not approved. The server is unchanged."
- id: end_declined
type: io.kestra.plugin.core.execution.Exit
state: SUCCESS
- id: recheck_pool
type: io.kestra.plugin.scripts.shell.Commands
description: Reads the pool again, because servers can change while the approval
waits. The capacity check runs against this read.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: ""
HA_TIMEOUT: "0"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: recheck_gate
type: io.kestra.plugin.core.flow.If
description: Refuses again if the approval is stale, for example after another
server went down.
condition: "{{ outputs.recheck_pool.vars.others_up < inputs.min_healthy or
outputs.recheck_pool.vars.found == false }}"
then:
- id: notify_refused_late
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
Drain of {{ render(vars.backend) }}/{{ render(vars.server) }} was
approved but refused and nothing was changed: the pool changed
while the approval waited. Now {{
outputs.recheck_pool.vars.others_up }} other server(s) are UP and
the minimum is {{ inputs.min_healthy }}. Check the servers, then
run again.
- id: refuse_late
type: io.kestra.plugin.core.execution.Exit
state: FAILED
- id: start_drain
type: io.kestra.plugin.scripts.shell.Commands
description: Sets DRAIN. HAProxy stops sending new sessions at once; the open
ones keep running.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_STATE: drain
commands:
- "{{ vars.set_state }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: wait_drain
type: io.kestra.plugin.core.flow.LoopUntil
description: Reads the session count every poll_seconds until it reaches 0, or
drain_timeout_seconds passes.
condition: "{{ outputs.poll_drain.vars.target_sessions == 0 or
outputs.poll_drain.vars.timed_out == true }}"
checkFrequency:
interval: "PT{{ inputs.poll_seconds }}S"
maxIterations: 1000
failOnMaxReached: false
tasks:
- id: poll_drain
type: io.kestra.plugin.scripts.shell.Commands
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: "{{ outputs.start_drain.vars.started_at }}"
HA_TIMEOUT: "{{ inputs.drain_timeout_seconds }}"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: confirm_drained
type: io.kestra.plugin.scripts.shell.Commands
description: One more read after the wait, so the MAINT decision uses the
current session count.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: ""
HA_TIMEOUT: "0"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: drain_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.confirm_drained.vars.target_sessions > 0 }}"
then:
- id: drain_timed_out
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Drain of {{ render(vars.backend) }}/{{ render(vars.server) }}
timed out after {{ inputs.drain_timeout_seconds }} s with {{
outputs.confirm_drained.vars.target_sessions }} session(s) still
open. The server stays in DRAIN: no new sessions, the open ones
continue. Nothing was set to MAINT. Find the slow clients, or run
with action enable to put the server back in rotation."
- id: set_maint
type: io.kestra.plugin.scripts.shell.Commands
description: Sets MAINT once no sessions are left. Safe because DRAIN has
stopped new sessions.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_STATE: maint
commands:
- "{{ vars.set_state }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: confirm_maint
type: io.kestra.plugin.scripts.shell.Commands
description: Reads the server back to confirm MAINT is what HAProxy now reports.
taskRunner:
type: io.kestra.plugin.core.runner.Process
interpreter:
- bash
- -c
env:
HA_HOST: "{{ inputs.haproxy_host }}"
HA_PORT: "{{ inputs.haproxy_port }}"
HA_BACKEND: "{{ render(vars.backend) }}"
HA_SERVER: "{{ render(vars.server) }}"
HA_START: ""
HA_TIMEOUT: "0"
commands:
- "{{ vars.read_stats }}"
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxAttempts: 3
- id: maint_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.confirm_maint.vars.target_status != 'MAINT' }}"
then:
- id: not_maint
type: io.kestra.plugin.core.execution.Fail
errorMessage: "{{ render(vars.backend) }}/{{ render(vars.server) }} was set to
MAINT but HAProxy reports status {{
outputs.confirm_maint.vars.target_status }}. Check the server with
`show servers state` on the runtime socket."
- id: notify_drained
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
{{ render(vars.backend) }}/{{ render(vars.server) }} is in
maintenance: its sessions drained to 0 and HAProxy reports {{
outputs.confirm_maint.vars.target_status }}. {{
outputs.recheck_pool.vars.others_up }} other server(s) are UP. When
the work is done, run this flow with action enable.
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
HAProxy drain gate failed in execution {{ execution.id }} {% if
errorLogs() | length > 0 %}at `{{ errorLogs()[0]['taskId'] }}`: {{
errorLogs()[0]['message'] | split(' \[\[') | first }}{% else %}without an
error message{% endif %}. The server state may have changed; check it in
HAProxy before you act. {{ inputs.kestra_url }}/ui/main/executions/{{
flow.namespace }}/{{ flow.id }}/{{ execution.id }}
outputs:
- id: server_status
type: STRING
description: HAProxy status of the server at the end of the run (MAINT after a
drain, UP after an enable). Empty when the run was refused, declined or
failed, because outputs are not written on those runs.
value: "{{ outputs.confirm_health.vars.target_status ??
outputs.confirm_maint.vars.target_status ?? '' }}"
