Jon Chery
5cbe3020d3
scripts/orca-aggregate.sh: 10s aggregator, cluster.json merge + drift-events rsync + remediation trigger (P10b stub). scripts/orca- watchdog.sh: C-11 starvation detection. internal/cli/collector.go: orca collector start/stop/status. Tests: CLI + bats. ---ci--- project: orca phase: 09 milestone: v0.11 status: execute ---/ci---
173 lines
6.8 KiB
Bash
Executable File
173 lines
6.8 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
# orca-aggregate.sh — lead-side aggregator (P09, C-11/C-12/C-14, REQ-107, D-237).
|
|
#
|
|
# Runs every 10s via orca-aggregate.timer. For each peer in
|
|
# /etc/orca/peers.env: SSHs in, reads /run/orca/state/<latest>.json (the
|
|
# peer's state snapshot), and merges all peer states into a single
|
|
# /etc/orca/state/cluster.json on the lead. Writes atomically (temp +
|
|
# rename). Structured logging via logger -t orca-aggregate (syslog).
|
|
#
|
|
# Drift-event aggregation extension (REQ-107, D-237): rsyncs each
|
|
# peer's /etc/orca/state/drift-events/ to a temp dir, validates each
|
|
# event's hash against /etc/orca/state/applied/<txn>/manifest.json,
|
|
# triggers orca-remediate.sh <peer> <txn> for auto-remediable paths
|
|
# (the remediation script itself lands in P10b — the call is emitted
|
|
# but the script may not exist yet), then consumes (deletes) the event
|
|
# file on the peer after processing. Aggregated drift state is written
|
|
# to /etc/orca/state/drift-events-aggregated.json. The drift section
|
|
# is skipped gracefully (with a log line) when the drift-events
|
|
# directory does not exist on the lead or on any peer.
|
|
#
|
|
# The aggregator is opt-in (C-12): the operator enables it with
|
|
# `orca cluster config --aggregator=true`, which installs this script
|
|
# + the systemd units via `orca collector start`.
|
|
|
|
set -euo pipefail
|
|
|
|
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
|
# shellcheck source=lib/orca-log.sh
|
|
. "$SCRIPT_DIR/lib/orca-log.sh"
|
|
|
|
ORCA_LOG_ACTOR="spiffe://orca/cli/aggregator"
|
|
|
|
PEERS_ENV="${ORCA_PEERS_ENV:-/etc/orca/peers.env}"
|
|
STATE_DIR="${ORCA_STATE_DIR:-/etc/orca/state}"
|
|
CLUSTER_JSON="$STATE_DIR/cluster.json"
|
|
DRIFT_AGG_JSON="$STATE_DIR/drift-events-aggregated.json"
|
|
REMOTE_STATE_DIR="${ORCA_REMOTE_STATE_DIR:-/run/orca/state}"
|
|
REMOTE_DRIFT_DIR="${ORCA_REMOTE_DRIFT_DIR:-/etc/orca/state/drift-events}"
|
|
APPLIED_DIR="${ORCA_APPLIED_DIR:-/etc/orca/state/applied}"
|
|
REMEDIATE_SCRIPT="${ORCA_REMEDIATE_SCRIPT:-/usr/local/sbin/orca-remediate.sh}"
|
|
SSH_OPTS="${ORCA_SSH_OPTS:--o BatchMode=yes -o StrictHostKeyChecking=accept-new -o ConnectTimeout=5}"
|
|
RSYNC_OPTS_RSYNC="${ORCA_RSYNC_OPTS_RSYNC:-}"
|
|
AGG_TMP="$(mktemp -d)"
|
|
trap 'rm -rf "$AGG_TMP"' EXIT
|
|
|
|
mkdir -p "$STATE_DIR"
|
|
|
|
read_peers() {
|
|
if [ ! -f "$PEERS_ENV" ]; then
|
|
return 0
|
|
fi
|
|
awk -F= '/^[[:space:]]*#/ {next} /^[[:space:]]*$/ {next} {sub(/^[[:space:]]*PEER_[^=]*=/,""); print}' "$PEERS_ENV"
|
|
}
|
|
|
|
merge_tmp="$AGG_TMP/merge.jsonl"
|
|
: >"$merge_tmp"
|
|
|
|
peer_count=0
|
|
while IFS= read -r peer; do
|
|
[ -z "$peer" ] && continue
|
|
peer_count=$((peer_count + 1))
|
|
latest_json="$(ssh $SSH_OPTS "$peer" "ls -1 $REMOTE_STATE_DIR/*.json 2>/dev/null | sort | tail -1" 2>/dev/null || true)"
|
|
if [ -z "$latest_json" ]; then
|
|
orca_log_warn "aggregate" "$peer" "skipped" "no state snapshot on peer"
|
|
continue
|
|
fi
|
|
snapshot="$(ssh $SSH_OPTS "$peer" "cat '$latest_json'" 2>/dev/null || true)"
|
|
if [ -z "$snapshot" ]; then
|
|
orca_log_warn "aggregate" "$peer" "skipped" "failed to read state snapshot"
|
|
continue
|
|
fi
|
|
printf '{"peer":"%s","state":%s}\n' "$peer" "$snapshot" >>"$merge_tmp"
|
|
orca_log_info "aggregate" "$peer" "ok" "snapshot=$latest_json"
|
|
done < <(read_peers)
|
|
|
|
ts="$(date -u +%Y-%m-%dT%H:%M:%S.%3NZ)"
|
|
{
|
|
printf '{"ts":"%s","peers":[' "$ts"
|
|
first=1
|
|
while IFS= read -r line; do
|
|
[ -z "$line" ] && continue
|
|
if [ "$first" -eq 1 ]; then
|
|
first=0
|
|
else
|
|
printf ','
|
|
fi
|
|
printf '%s' "$line"
|
|
done <"$merge_tmp"
|
|
printf ']}'
|
|
} >"$AGG_TMP/cluster.json.new"
|
|
mv "$AGG_TMP/cluster.json.new" "$CLUSTER_JSON"
|
|
orca_log_info "aggregate" "-" "ok" "peers=$peer_count cluster_json=$CLUSTER_JSON"
|
|
|
|
if [ ! -d "$REMOTE_DRIFT_DIR" ] && [ ! -d "$APPLIED_DIR" ]; then
|
|
orca_log_info "drift" "-" "skipped" "drift-events + applied dirs absent (P10b not shipped)"
|
|
else
|
|
drift_tmp="$AGG_TMP/drift-events"
|
|
mkdir -p "$drift_tmp"
|
|
drift_agg_tmp="$AGG_TMP/drift-agg.jsonl"
|
|
: >"$drift_agg_tmp"
|
|
while IFS= read -r peer; do
|
|
[ -z "$peer" ] && continue
|
|
peer_drift_dir="$drift_tmp/$peer"
|
|
mkdir -p "$peer_drift_dir"
|
|
if ! rsync -a --quiet $RSYNC_OPTS_RSYNC "$peer:$REMOTE_DRIFT_DIR/" "$peer_drift_dir/" 2>/dev/null; then
|
|
orca_log_info "drift" "$peer" "skipped" "rsync failed or dir absent"
|
|
continue
|
|
fi
|
|
shopt -s nullglob
|
|
event_files=("$peer_drift_dir"/*.json)
|
|
shopt -u nullglob
|
|
if [ "${#event_files[@]}" -eq 0 ]; then
|
|
orca_log_info "drift" "$peer" "ok" "no events"
|
|
continue
|
|
fi
|
|
for event_file in "${event_files[@]}"; do
|
|
event_id="$(basename "$event_file" .json)"
|
|
path_val="$(grep -o '"path"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
|
|
new_sha="$(grep -o '"new_sha256"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
|
|
txn_val="$(grep -o '"latest_txn"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
|
|
status_val="$(grep -o '"status"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
|
|
if [ -z "$txn_val" ]; then
|
|
orca_log_warn "drift" "$peer" "skipped" "event=$event_id no latest_txn"
|
|
continue
|
|
fi
|
|
manifest="$APPLIED_DIR/$txn_val/manifest.json"
|
|
drift_confirmed=0
|
|
if [ -f "$manifest" ]; then
|
|
manifest_sha="$(grep -o "\"$path_val\"[^\"]*\"sha256\"[[:space:]]*:[[:space:]]*\"[^\"]*\"" "$manifest" 2>/dev/null | grep -o '"sha256"[[:space:]]*:[[:space:]]*"[^"]*"' | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
|
|
if [ -n "$manifest_sha" ] && [ "$manifest_sha" != "$new_sha" ]; then
|
|
drift_confirmed=1
|
|
fi
|
|
else
|
|
orca_log_warn "drift" "$peer" "skipped" "txn=$txn_val manifest absent"
|
|
fi
|
|
if [ "$drift_confirmed" -eq 1 ]; then
|
|
if [ -x "$REMEDIATE_SCRIPT" ]; then
|
|
orca_log_info "drift" "$peer" "remediate" "txn=$txn_val path=$path_val"
|
|
"$REMEDIATE_SCRIPT" "$peer" "$txn_val" "$path_val" \
|
|
|| orca_log_error "drift" "$peer" "remediate-failed" "txn=$txn_val path=$path_val"
|
|
else
|
|
orca_log_warn "drift" "$peer" "remediate-stub" "orca-remediate.sh not installed (P10b)"
|
|
fi
|
|
printf '{"peer":"%s","event_id":"%s","txn":"%s","path":"%s","status":"%s","remediated":true}\n' \
|
|
"$peer" "$event_id" "$txn_val" "$path_val" "$status_val" >>"$drift_agg_tmp"
|
|
else
|
|
printf '{"peer":"%s","event_id":"%s","txn":"%s","path":"%s","status":"%s","remediated":false}\n' \
|
|
"$peer" "$event_id" "$txn_val" "$path_val" "$status_val" >>"$drift_agg_tmp"
|
|
fi
|
|
ssh $SSH_OPTS "$peer" "rm -f '$REMOTE_DRIFT_DIR/$event_id.json'" 2>/dev/null \
|
|
|| orca_log_warn "drift" "$peer" "consume-failed" "event=$event_id"
|
|
done
|
|
done < <(read_peers)
|
|
{
|
|
printf '{"ts":"%s","events":[' "$ts"
|
|
first=1
|
|
while IFS= read -r line; do
|
|
[ -z "$line" ] && continue
|
|
if [ "$first" -eq 1 ]; then
|
|
first=0
|
|
else
|
|
printf ','
|
|
fi
|
|
printf '%s' "$line"
|
|
done <"$drift_agg_tmp"
|
|
printf ']}'
|
|
} >"$AGG_TMP/drift-agg.json.new"
|
|
mv "$AGG_TMP/drift-agg.json.new" "$DRIFT_AGG_JSON"
|
|
orca_log_info "drift" "-" "ok" "aggregated=$DRIFT_AGG_JSON"
|
|
fi
|
|
|
|
orca_log_info "aggregate" "-" "done" "peers=$peer_count"
|