376 lines
14 KiB
Bash
376 lines
14 KiB
Bash
#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
|
|
# --- Config (override via env) ---
|
|
RMQ_USER="${RMQ_USER:-admin}"
|
|
RMQ_PASS="${RMQ_PASS:-change_me}"
|
|
RMQ_HOST="${RMQ_HOST:-localhost}"
|
|
RMQ_PORT="${RMQ_PORT:-15672}"
|
|
VHOST="${VHOST:-app}"
|
|
|
|
# Hard override as in your current file
|
|
RMQ_HOST="10.210.12.2"
|
|
|
|
# Mode A: read from a queue (default). Supports CSV for multiple queues.
|
|
# CHANGED: default now includes both normal and persuasive work queues.
|
|
QUEUE="${QUEUE:-queue_deviceconfig,queue_persuasive}" # e.g. "queue_deviceconfig,queue_persuasive"
|
|
|
|
# Mode B: bind temp queue to an exchange + routing key and consume from it
|
|
EXCHANGE="${EXCHANGE:-}" # empty => Mode B OFF
|
|
ROUTING_KEY="${ROUTING_KEY:-}" # irrelevant when EXCHANGE is empty
|
|
|
|
# Polling interval when no messages
|
|
SLEEP_SECS="${SLEEP_SECS:-1}"
|
|
|
|
# GO/NO-GO gate bypass (requested): set ignore_gonogo=true to ignore gate
|
|
ignore_gonogo="${ignore_gonogo:-false}"
|
|
|
|
# Paths
|
|
APP_ROOT="/opt/containers/ansible-worker/app"
|
|
NBPLAY="${APP_ROOT}/bin/nbplay"
|
|
|
|
# --- Helpers ---
|
|
api() {
|
|
local method="$1"; shift
|
|
local path="$1"; shift
|
|
local data="${1:-}"
|
|
if [[ -n "$data" ]]; then
|
|
curl -sS -u "$RMQ_USER:$RMQ_PASS" -H "content-type:application/json" -X "$method" "http://$RMQ_HOST:$RMQ_PORT$path" -d "$data"
|
|
else
|
|
curl -sS -u "$RMQ_USER:$RMQ_PASS" -H "content-type:application/json" -X "$method" "http://$RMQ_HOST:$RMQ_PORT$path"
|
|
fi
|
|
}
|
|
|
|
has_jq() { command -v jq >/dev/null 2>&1; }
|
|
|
|
urlenc() { printf '%s' "$1" | sed -e 's, ,%20,g' -e 's,/,%2F,g'; }
|
|
|
|
log() { printf '[consumer] %s\n' "$*"; }
|
|
warn() { printf '[consumer][WARN] %s\n' "$*" >&2; }
|
|
err() { printf '[consumer][ERROR] %s\n' "$*" >&2; }
|
|
|
|
# --- GO/NO-GO gate helpers (surgical add) ---
|
|
gate_sleep_secs() {
|
|
# Random 2..5 seconds inclusive
|
|
echo $(( (RANDOM % 4) + 2 ))
|
|
}
|
|
|
|
gate_is_go() {
|
|
# Bypass if ignore_gonogo is true-ish
|
|
case "${ignore_gonogo,,}" in
|
|
true|1|yes|y) return 0 ;;
|
|
esac
|
|
|
|
# Fail-closed: any error/unreachable => NO-GO
|
|
local body val
|
|
if ! body="$(curl -fsS --connect-timeout 2 --max-time 3 "http://10.210.12.2:8090/data/go_nogo.txt" 2>/dev/null)"; then
|
|
return 1
|
|
fi
|
|
|
|
# Must match ^go$ (allow trailing newline in file)
|
|
val="$(printf '%s' "$body" | tr -d '\r' | head -n1 | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//')"
|
|
val="${val,,}"
|
|
[[ "$val" == "go" ]]
|
|
}
|
|
|
|
gate_block_if_needed() {
|
|
# If gate is closed, sleep random 2..5 seconds and signal caller to continue loop
|
|
if gate_is_go; then
|
|
return 0
|
|
fi
|
|
sleep "$(gate_sleep_secs)"
|
|
return 1
|
|
}
|
|
|
|
# Safely append string options into an array using eval (so quotes are honored).
|
|
append_opts() {
|
|
local opts_str="$1"
|
|
# shellcheck disable=SC2206
|
|
local -n _arr_ref=$2
|
|
if [[ -n "$opts_str" ]]; then
|
|
eval '_arr_ref+=('"$opts_str"')'
|
|
fi
|
|
}
|
|
|
|
# Given a JSON object payload, extract fields and dispatch nbplay
|
|
dispatch_task() {
|
|
local json="$1"
|
|
local device task playbook task_options
|
|
device="$(jq -er '.inscope_device // empty' <<<"$json")" || device=""
|
|
task="$(jq -er '.task_name // empty' <<<"$json")" || task=""
|
|
task_options="$(jq -r '.task_options // empty' <<<"$json")" || task_options=""
|
|
|
|
# Pass after-upgrade metadata via -e by augmenting task_options (single source of truth)
|
|
# Supports both historic 'afterupgrade_check' and current 'afterupgrade_indoor_check'
|
|
if [[ "$task" == "afterupgrade_indoor_check" || "$task" == "afterupgrade_check" ]]; then
|
|
# Extract everything the publisher may send
|
|
local attempt max_attempts cur_delay corr_id orig_at target_ver tv_full schema
|
|
attempt="$(jq -r '.attempt // ""' <<<"$json")"
|
|
max_attempts="$(jq -r '.max_attempts // ""' <<<"$json")"
|
|
cur_delay="$(jq -r '.current_delay_sec // ""' <<<"$json")"
|
|
corr_id="$(jq -r '.correlation_id // ""' <<<"$json")"
|
|
orig_at="$(jq -r '.original_emitted_at // ""' <<<"$json")"
|
|
target_ver="$(jq -r '.target_version // ""' <<<"$json")"
|
|
tv_full="$(jq -r '.target_version_full // ""' <<<"$json")"
|
|
schema="$(jq -r '.schema_version // ""' <<<"$json")"
|
|
is_run_by="$(jq -r '.is_run_by // ""' <<<"$json")"
|
|
|
|
# Backfill target_version from target_version_full if missing (e.g., fox200-2.2.1-r6801.bin → 2.2.1-r6801)
|
|
if [[ -z "$target_ver" && -n "$tv_full" ]]; then
|
|
target_ver="$(sed -nE 's/.*([0-9]+\.[0-9]+\.[0-9]+-r[0-9]+).*/\1/p' <<<"$tv_full" || true)"
|
|
fi
|
|
|
|
# Safe single-quote escaper for -e 'value'
|
|
esc() { local s="$1"; printf "%s" "${s//\'/\047}"; }
|
|
|
|
# Always pass the vars (even if empty), so the play never sees undefined
|
|
task_options+=" -e attempt='$(esc "$attempt")'"
|
|
task_options+=" -e max_attempts='$(esc "$max_attempts")'"
|
|
task_options+=" -e current_delay_sec='$(esc "$cur_delay")'"
|
|
task_options+=" -e correlation_id='$(esc "$corr_id")'"
|
|
task_options+=" -e original_emitted_at='$(esc "$orig_at")'"
|
|
task_options+=" -e target_version='$(esc "$target_ver")'"
|
|
task_options+=" -e target_version_full='$(esc "$tv_full")'"
|
|
task_options+=" -e schema_version='$(esc "$schema")'"
|
|
task_options+=" -e is_run_by='$(esc "$is_run_by")'"
|
|
fi
|
|
|
|
if [[ -z "$device" || -z "$task" ]]; then
|
|
warn "payload missing required keys (inscope_device/task_name). Skipping."
|
|
return 0
|
|
fi
|
|
|
|
# --- Special cases: parametric reboot hours from task name
|
|
local base_task="$task"
|
|
local -a extra_nbplay_opts=()
|
|
local n=""
|
|
|
|
# NEW: allow trailing "_force" suffix (can be combined with other suffixes like _tonight / _<hours>)
|
|
if [[ "$base_task" =~ ^(.+)_force$ ]]; then
|
|
base_task="${BASH_REMATCH[1]}"
|
|
extra_nbplay_opts+=("-eforce_upgrade=yes")
|
|
log "Parsed _force suffix: base='${base_task}' (passed as -e force_upgrade=yes)"
|
|
fi
|
|
|
|
# NEW: generic *_tonight → compute hours until next 01:00 (ceil) + random 1..4
|
|
if [[ "$base_task" =~ ^(.+)_tonight$ ]]; then
|
|
base_task="${BASH_REMATCH[1]}"
|
|
|
|
# now, today 01:00, tomorrow 01:00 (local time)
|
|
local now_s today1_s tomorrow1_s next1_s diff_s ceil_h rnd extra_h total_h
|
|
now_s="$(date +%s)"
|
|
today1_s="$(date -d 'today 00:00' +%s)"
|
|
tomorrow1_s="$(date -d 'tomorrow 00:00' +%s)"
|
|
if (( now_s < today1_s )); then
|
|
next1_s="$today1_s"
|
|
else
|
|
next1_s="$tomorrow1_s"
|
|
fi
|
|
diff_s=$(( next1_s - now_s ))
|
|
# Ceil hours so current minutes are preserved as in your examples
|
|
ceil_h=$(( (diff_s + 3599) / 3600 ))
|
|
rnd=$(( (RANDOM % 4) + 1 )) # 1..4
|
|
total_h=$(( ceil_h + rnd - 1 ))
|
|
|
|
# NEW: if rebootin is huge (>=18h), convert to immediate reboot (0h)
|
|
if (( total_h >= 19 )); then
|
|
total_h=0
|
|
log "Adjusted _tonight rebootin to 0h because computed value was >=18h"
|
|
fi
|
|
|
|
extra_nbplay_opts+=("-erebootin=${total_h}")
|
|
log "Resolved '${task}' → base='${base_task}', rebootin=${total_h}h (ceil_to_1am=${ceil_h}h + rand=${rnd}h)"
|
|
elif [[ "$base_task" =~ ^update-rebootin_([0-9]{1,2})$ ]]; then
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-rebootin"
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
elif [[ "$base_task" =~ ^update-reboot_([0-9]{1,4})$ ]]; then
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-reboot" # runs update-reboot.yml (wrapper -> update-rebootin222.yml)
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
elif [[ "$base_task" =~ ^update-reboot-scheduler_([0-9]{1,4})$ ]]; then
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-reboot-scheduler"
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
elif [[ "$base_task" =~ ^(update-reboot[0-9]{3}-scheduler)_([0-9]{1,4})$ ]]; then
|
|
base_task="${BASH_REMATCH[1]}"
|
|
n="${BASH_REMATCH[2]}"
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
elif [[ "$base_task" =~ ^update-indoor_([0-9]{1,4})$ ]]; then
|
|
# CHANGED: keep same convention as others — pass HOURS directly via -e rebootin=<n>
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-indoor"
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
log "Parsed update-indoor suffix: ${n}h (passed as -e rebootin=${n})"
|
|
elif [[ "$base_task" =~ ^update-indoor-bootenv_([0-9]{1,4})$ ]]; then
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-indoor-bootenv"
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
log "Parsed update-indoor-bootenv suffix: ${n}h (passed as -e rebootin=${n})"
|
|
fi
|
|
|
|
# --- Default rebootin for reboot-family when not explicitly provided ---
|
|
if [[ "$base_task" =~ ^(update-rebootin222|update-rebootin|update-reboot|update-reboot-scheduler|update-reboot[0-9]{3}-scheduler)$ ]]; then
|
|
# only set if neither task_options nor extra_nbplay_opts already contain rebootin
|
|
if [[ "$task_options" != *"rebootin="* ]] && ! printf '%s\n' "${extra_nbplay_opts[@]}" | grep -q 'rebootin='; then
|
|
# restore legacy behavior: immediate reboot if none specified
|
|
extra_nbplay_opts+=("-erebootin=0")
|
|
log "No rebootin provided; defaulting to rebootin=0 for ${base_task}"
|
|
fi
|
|
fi
|
|
|
|
playbook="${APP_ROOT}/${base_task}.yml"
|
|
if [[ ! -f "$playbook" ]]; then
|
|
warn "playbook not found: $playbook (device='$device', task='$task'). Skipping."
|
|
return 0
|
|
fi
|
|
|
|
log "→ Running $(basename "$playbook") on '${device}' with options: ${task_options:-<none>}"
|
|
|
|
local -a cmd=( "$NBPLAY" "$playbook" "$device" )
|
|
append_opts "$task_options" cmd
|
|
|
|
if ((${#extra_nbplay_opts[@]})); then
|
|
cmd+=("${extra_nbplay_opts[@]}")
|
|
fi
|
|
|
|
# Ensure playbooks publish control/journal/tag to the correct exchange without requiring container env.
|
|
# This sets RMQ_EXCHANGE=controls only for this nbplay invocation.
|
|
if ! RMQ_EXCHANGE=controls "${cmd[@]}"; then
|
|
err "playbook failed (task='${task}', device='${device}')"
|
|
# keep consuming; message already acked by HTTP get endpoint
|
|
fi
|
|
}
|
|
|
|
cleanup_queue=""
|
|
cleanup() {
|
|
if [[ -n "$cleanup_queue" ]]; then
|
|
log "Cleaning up temp queue: $cleanup_queue"
|
|
api DELETE "/api/queues/$(urlenc "$VHOST")/$cleanup_queue" >/dev/null || true
|
|
fi
|
|
}
|
|
trap cleanup EXIT
|
|
|
|
# --- Setup: decide source ---
|
|
if [[ -n "$EXCHANGE" ]]; then
|
|
# Mode B (unchanged): bind a temp queue to exchange+routing key
|
|
cleanup_queue="tmp.$(hostname -s).$$.$(date +%s)"
|
|
log "Declaring temp queue: $cleanup_queue (exclusive, auto-delete)"
|
|
api PUT "/api/queues/$(urlenc "$VHOST")/$cleanup_queue" '{
|
|
"auto_delete": true, "durable": false, "arguments": {}, "exclusive": true
|
|
}' >/dev/null
|
|
|
|
log "Binding temp queue to exchange '$EXCHANGE' with routing key '$ROUTING_KEY'"
|
|
api POST "/api/bindings/$(urlenc "$VHOST")/e/$EXCHANGE/q/$cleanup_queue" "{
|
|
\"routing_key\": \"$ROUTING_KEY\", \"arguments\": {}
|
|
}" >/dev/null
|
|
|
|
QUEUE="$cleanup_queue"
|
|
log "Consuming from bound temp queue: $QUEUE"
|
|
|
|
# Single-queue consume loop (unchanged branch)
|
|
log "Press Ctrl+C to stop."
|
|
while :; do
|
|
# GO/NO-GO gate: do NOT dequeue unless gate is GO
|
|
if ! gate_block_if_needed; then
|
|
continue
|
|
fi
|
|
|
|
RESP="$(api POST "/api/queues/$(urlenc "$VHOST")/$QUEUE/get" '{
|
|
"count": 1, "ackmode": "ack_requeue_false", "encoding": "auto", "truncate": 1000000
|
|
}')"
|
|
|
|
if [[ "$RESP" == "[]" || -z "$RESP" ]]; then
|
|
sleep "$SLEEP_SECS"
|
|
continue
|
|
fi
|
|
|
|
if has_jq; then
|
|
echo "$RESP" | jq -c '.[]' | while read -r item; do
|
|
payload=$(printf '%s' "$item" | jq -r '.payload')
|
|
if jq -e . >/dev/null 2>&1 <<<"$payload"; then
|
|
ptype="$(printf '%s' "$payload" | jq -r 'type')"
|
|
if [[ "$ptype" == "string" ]]; then
|
|
decoded="$(printf '%s' "$payload" | jq -r .)"
|
|
if jq -e . >/dev/null 2>&1 <<<"$decoded"; then
|
|
dispatch_task "$decoded"
|
|
else
|
|
warn "payload is a string but not JSON after unquote; skipping."
|
|
fi
|
|
else
|
|
dispatch_task "$payload"
|
|
fi
|
|
else
|
|
warn "payload is not JSON; skipping."
|
|
fi
|
|
done
|
|
else
|
|
warn "jq not found; cannot parse JSON payloads. Exiting."
|
|
exit 3
|
|
fi
|
|
done
|
|
|
|
else
|
|
# Mode A: direct queue(s). Support CSV list in QUEUE.
|
|
IFS=',' read -r -a QUEUE_LIST <<< "$QUEUE"
|
|
for i in "${!QUEUE_LIST[@]}"; do QUEUE_LIST[$i]="${QUEUE_LIST[$i]//[[:space:]]/}"; done
|
|
|
|
if [[ ${#QUEUE_LIST[@]} -eq 0 ]]; then
|
|
err "No queues configured (QUEUE env empty)."
|
|
exit 2
|
|
fi
|
|
|
|
log "Consuming from queue(s): ${QUEUE_LIST[*]} (vhost: $VHOST)"
|
|
log "Press Ctrl+C to stop."
|
|
|
|
# Multi-queue round-robin: try each queue once per loop; if any yields a message, process it and start over.
|
|
while :; do
|
|
# GO/NO-GO gate: do NOT dequeue unless gate is GO
|
|
if ! gate_block_if_needed; then
|
|
continue
|
|
fi
|
|
|
|
local_got_message=0
|
|
|
|
for Q in "${QUEUE_LIST[@]}"; do
|
|
RESP="$(api POST "/api/queues/$(urlenc "$VHOST")/$Q/get" '{
|
|
"count": 1, "ackmode": "ack_requeue_false", "encoding": "auto", "truncate": 1000000
|
|
}')"
|
|
|
|
if [[ "$RESP" != "[]" && -n "$RESP" ]]; then
|
|
local_got_message=1
|
|
log "Dequeued from [$Q]"
|
|
if has_jq; then
|
|
echo "$RESP" | jq -c '.[]' | while read -r item; do
|
|
payload=$(printf '%s' "$item" | jq -r '.payload')
|
|
if jq -e . >/dev/null 2>&1 <<<"$payload"; then
|
|
ptype="$(printf '%s' "$payload" | jq -r 'type')"
|
|
if [[ "$ptype" == "string" ]]; then
|
|
decoded="$(printf '%s' "$payload" | jq -r .)"
|
|
if jq -e . >/dev/null 2>&1 <<<"$decoded"; then
|
|
dispatch_task "$decoded"
|
|
else
|
|
warn "payload is a string but not JSON after unquote; skipping."
|
|
fi
|
|
else
|
|
dispatch_task "$payload"
|
|
fi
|
|
else
|
|
warn "payload is not JSON; skipping."
|
|
fi
|
|
done
|
|
else
|
|
warn "jq not found; cannot parse JSON payloads. Exiting."
|
|
exit 3
|
|
fi
|
|
|
|
# After processing a message from this queue, start the round over (fair-ish polling).
|
|
break
|
|
fi
|
|
done
|
|
|
|
if [[ "$local_got_message" -eq 0 ]]; then
|
|
sleep "$SLEEP_SECS"
|
|
fi
|
|
done
|
|
fi |