256 lines
8.7 KiB
Bash
256 lines
8.7 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}"
|
|
|
|
# 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; }
|
|
|
|
# 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)
|
|
if [[ "$task" == "afterupgrade_check" ]]; then
|
|
local attempt corr_id orig_at target_ver
|
|
attempt="$(jq -r '.attempt // empty' <<<"$json")"
|
|
corr_id="$(jq -r '.correlation_id // empty' <<<"$json")"
|
|
orig_at="$(jq -r '.original_emitted_at // empty' <<<"$json")"
|
|
target_ver="$(jq -r '.target_version // empty' <<<"$json")"
|
|
|
|
[[ -n "$attempt" ]] && task_options+=" -e attempt=${attempt}"
|
|
[[ -n "$corr_id" ]] && task_options+=" -e correlation_id=${corr_id}"
|
|
[[ -n "$orig_at" ]] && task_options+=" -e original_emitted_at=${orig_at}"
|
|
|
|
if [[ -n "$target_ver" ]]; then
|
|
local esc_tv=${target_ver//\'/\'\"\'\"\'} # replace ' with '\'' safely
|
|
task_options+=" -e target_version='${esc_tv}'"
|
|
fi
|
|
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=""
|
|
if [[ "$task" =~ ^update-rebootin_([0-9]{1,2})$ ]]; then
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-rebootin"
|
|
extra_nbplay_opts+=("-erebootin=${n}")
|
|
elif [[ "$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 [[ "$task" =~ ^update-indoor_([0-9]{1,4})$ ]]; then
|
|
n="${BASH_REMATCH[1]}"
|
|
base_task="update-indoor"
|
|
# Convert hours -> minutes for nbplay's --rebootin
|
|
minutes=$((10#$n * 60))
|
|
extra_nbplay_opts+=("--rebootin" "${minutes}")
|
|
log "Parsed update-indoor suffix: ${n}h -> ${minutes}m (rebootin)"
|
|
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
|
|
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
|
|
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
|