first commit
This commit is contained in:
ansible user
241
files/rabbit-client.sh-beforequeuefixing
Normal file
241
files/rabbit-client.sh-beforequeuefixing
Normal file
@@ -0,0 +1,241 @@
|
||||
#!/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.
|
||||
QUEUE="${QUEUE:-queue_deviceconfig}" # 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 case: task "update-rebootin_N" (N=0..99)
|
||||
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}")
|
||||
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
|
||||
|
||||
if ! "${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[@]} == 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
|
||||
Reference in New Issue
Block a user