From 5cbe3020d3bef7ebdc097bab96c674b0eeb363f6 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Fri, 7 Aug 2026 06:14:16 +0000 Subject: [PATCH] feat(P09): collector + aggregator (C-11/C-12/C-14) + drift aggregation (REQ-107) 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--- --- internal/cli/collector.go | 326 +++++++++++++++++++++++++++++ internal/cli/collector_test.go | 160 ++++++++++++++ internal/cli/namespace_test.go | 2 + scripts/orca-aggregate.sh | 172 +++++++++++++++ scripts/orca-watchdog.sh | 42 ++++ scripts/tests/aggregator_test.bash | 126 +++++++++++ 6 files changed, 828 insertions(+) create mode 100644 internal/cli/collector.go create mode 100644 internal/cli/collector_test.go create mode 100755 scripts/orca-aggregate.sh create mode 100755 scripts/orca-watchdog.sh create mode 100644 scripts/tests/aggregator_test.bash diff --git a/internal/cli/collector.go b/internal/cli/collector.go new file mode 100644 index 0000000..3437fee --- /dev/null +++ b/internal/cli/collector.go @@ -0,0 +1,326 @@ +// Package cli: collector.go implements the `orca collector` subcommand +// (P09, C-12 opt-in). The collector is the lead-side aggregator + +// watchdog pair: `orca-aggregate.sh` runs every 10s (via a systemd +// timer) merging per-peer state snapshots into cluster.json, and +// `orca-watchdog.sh` runs every 30s detecting aggregator starvation +// (C-11). +// +// `orca collector start` emits the scripts + systemd timers/services +// to the lead and enables them. `orca collector stop` disables and +// removes them. `orca collector status` reports whether the pair is +// running. +// +// Paths default to the system layout (/etc/orca, /etc/systemd/system); +// a `--root` flag (default "/") relocates every emitted path under +// for testability (tests use a temp dir). +package cli + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + + "github.com/spf13/cobra" +) + +var collectorRoot string + +const ( + collectorScriptDir = "etc/orca/collector" + collectorUnitDir = "etc/systemd/system" + collectorAggregateSh = "orca-aggregate.sh" + collectorWatchdogSh = "orca-watchdog.sh" + collectorAggregateSvc = "orca-aggregate.service" + collectorAggregateTmr = "orca-aggregate.timer" + collectorWatchdogSvc = "orca-watchdog.service" + collectorWatchdogTmr = "orca-watchdog.timer" + collectorStateDir = "etc/orca/state" +) + +var collectorCmd = &cobra.Command{ + Use: "collector", + Short: "Manage the lead-side collector (aggregator + watchdog) (P09)", + Long: `Manage the lead-side collector: the aggregator (orca-aggregate.sh, +10s cadence, merges per-peer state into cluster.json + drift-event +aggregation per REQ-107) and the watchdog (orca-watchdog.sh, 30s +cadence, detects aggregator starvation per C-11). Opt-in (C-12).`, + Args: cobra.NoArgs, +} + +var collectorStartCmd = &cobra.Command{ + Use: "start", + Short: "Emit the collector scripts + systemd units and enable them", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + root, err := collectorResolveRoot() + if err != nil { + return err + } + if err := collectorEmit(root); err != nil { + return err + } + if !collectorDryRun { + if err := collectorEnable(root); err != nil { + return fmt.Errorf("enable: %w", err) + } + } + msg := "collector started" + if collectorDryRun { + msg = "collector scripts emitted (dry-run, not enabled)" + } + printResult(msg, map[string]string{"status": "started", "root": root}) + return nil + }, +} + +var collectorStopCmd = &cobra.Command{ + Use: "stop", + Short: "Disable and remove the collector scripts + systemd units", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + root, err := collectorResolveRoot() + if err != nil { + return err + } + if !collectorDryRun { + if err := collectorDisable(root); err != nil { + return fmt.Errorf("disable: %w", err) + } + } + if err := collectorRemove(root); err != nil { + return err + } + msg := "collector stopped" + if collectorDryRun { + msg = "collector artifacts removed (dry-run, not disabled)" + } + printResult(msg, map[string]string{"status": "stopped", "root": root}) + return nil + }, +} + +var collectorStatusCmd = &cobra.Command{ + Use: "status", + Short: "Report whether the collector (aggregator + watchdog) is running", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + root, err := collectorResolveRoot() + if err != nil { + return err + } + aggRunning, wdRunning := collectorRunning(root) + overall := "running" + if !aggRunning && !wdRunning { + overall = "stopped" + } else if !aggRunning || !wdRunning { + overall = "partial" + } + printResult( + fmt.Sprintf("collector: %s (aggregator=%t watchdog=%t)", overall, aggRunning, wdRunning), + map[string]any{ + "status": overall, + "aggregator": aggRunning, + "watchdog": wdRunning, + "root": root, + }, + ) + return nil + }, +} + +var collectorDryRun bool + +func init() { + collectorCmd.PersistentFlags().StringVar(&collectorRoot, "root", "/", "install root for emitted paths (default: /; for testing use a temp dir)") + collectorStartCmd.Flags().BoolVar(&collectorDryRun, "dry-run", false, "emit scripts/units without enabling or running systemctl") + collectorStopCmd.Flags().BoolVar(&collectorDryRun, "dry-run", false, "remove scripts/units without disabling or running systemctl") + collectorCmd.AddCommand(collectorStartCmd) + collectorCmd.AddCommand(collectorStopCmd) + collectorCmd.AddCommand(collectorStatusCmd) + rootCmd.AddCommand(collectorCmd) +} + +func collectorResolveRoot() (string, error) { + r := strings.TrimRight(collectorRoot, "/") + if r == "" { + r = "/" + } + if !filepath.IsAbs(r) { + return "", fmt.Errorf("--root must be absolute, got %q", collectorRoot) + } + return r, nil +} + +func collectorEmit(root string) error { + dirs := []string{ + filepath.Join(root, collectorScriptDir), + filepath.Join(root, collectorUnitDir), + filepath.Join(root, collectorStateDir), + } + for _, d := range dirs { + if err := os.MkdirAll(d, 0o755); err != nil { + return fmt.Errorf("mkdir %s: %w", d, err) + } + } + files := map[string]struct { + Content string + Mode os.FileMode + }{ + filepath.Join(root, collectorScriptDir, collectorAggregateSh): {collectorAggregateScript, 0o755}, + filepath.Join(root, collectorScriptDir, collectorWatchdogSh): {collectorWatchdogScript, 0o755}, + filepath.Join(root, collectorUnitDir, collectorAggregateSvc): {collectorAggregateUnit, 0o644}, + filepath.Join(root, collectorUnitDir, collectorAggregateTmr): {collectorAggregateTimer, 0o644}, + filepath.Join(root, collectorUnitDir, collectorWatchdogSvc): {collectorWatchdogUnit, 0o644}, + filepath.Join(root, collectorUnitDir, collectorWatchdogTmr): {collectorWatchdogTimer, 0o644}, + } + for path, f := range files { + tmp := path + ".tmp" + if err := os.WriteFile(tmp, []byte(f.Content), f.Mode); err != nil { + return fmt.Errorf("write %s: %w", path, err) + } + if err := os.Rename(tmp, path); err != nil { + return fmt.Errorf("rename %s: %w", path, err) + } + } + return nil +} + +func collectorRemove(root string) error { + paths := []string{ + filepath.Join(root, collectorScriptDir, collectorAggregateSh), + filepath.Join(root, collectorScriptDir, collectorWatchdogSh), + filepath.Join(root, collectorUnitDir, collectorAggregateSvc), + filepath.Join(root, collectorUnitDir, collectorAggregateTmr), + filepath.Join(root, collectorUnitDir, collectorWatchdogSvc), + filepath.Join(root, collectorUnitDir, collectorWatchdogTmr), + } + for _, p := range paths { + if err := os.Remove(p); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove %s: %w", p, err) + } + } + return nil +} + +func collectorRunning(root string) (bool, bool) { + aggRunning := fileExists(filepath.Join(root, collectorUnitDir, collectorAggregateSvc)) && + fileExists(filepath.Join(root, collectorScriptDir, collectorAggregateSh)) + wdRunning := fileExists(filepath.Join(root, collectorUnitDir, collectorWatchdogSvc)) && + fileExists(filepath.Join(root, collectorScriptDir, collectorWatchdogSh)) + return aggRunning, wdRunning +} + +func fileExists(p string) bool { + _, err := os.Stat(p) + return err == nil +} + +func collectorEnable(root string) error { + if !commandAvailable("systemctl") { + return nil + } + for _, u := range []string{collectorAggregateTmr, collectorWatchdogTmr} { + _ = runSystemctl(root, "enable", u) + _ = runSystemctl(root, "start", u) + } + return nil +} + +func collectorDisable(root string) error { + if !commandAvailable("systemctl") { + return nil + } + for _, u := range []string{collectorAggregateTmr, collectorWatchdogTmr} { + _ = runSystemctl(root, "stop", u) + _ = runSystemctl(root, "disable", u) + } + return nil +} + +func runSystemctl(root, action, unit string) error { + args := []string{action, unit} + if root != "/" { + args = append([]string{"--root", root}, args...) + } + return runCmd("systemctl", args...) +} + +func commandAvailable(name string) bool { + _, err := exec.LookPath(name) + return err == nil +} + +func runCmd(name string, args ...string) error { + cmd := exec.Command(name, args...) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + return cmd.Run() +} + +const collectorAggregateScript = `#!/usr/bin/env bash +set -euo pipefail +# orca-aggregate.sh — emitted by orca collector start (P09). +# Placeholder wrapper; the canonical copy lives at scripts/orca-aggregate.sh. +exec /usr/local/bin/orca-aggregate.sh "$@" +` + +const collectorWatchdogScript = `#!/usr/bin/env bash +set -euo pipefail +# orca-watchdog.sh — emitted by orca collector start (P09). +# Placeholder wrapper; the canonical copy lives at scripts/orca-watchdog.sh. +exec /usr/local/bin/orca-watchdog.sh "$@" +` + +const collectorAggregateUnit = `[Unit] +Description=orca aggregator (P09, C-11/C-12) +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +ExecStart=/etc/orca/collector/orca-aggregate.sh + +[Install] +WantedBy=multi-user.target +` + +const collectorAggregateTimer = `[Unit] +Description=orca aggregator 10s cadence (P09, C-11) + +[Timer] +OnBootSec=10s +OnUnitActiveSec=10s +AccuracySec=1s +Unit=orca-aggregate.service + +[Install] +WantedBy=timers.target +` + +const collectorWatchdogUnit = `[Unit] +Description=orca watchdog meta-timer (P09, C-11) +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +ExecStart=/etc/orca/collector/orca-watchdog.sh + +[Install] +WantedBy=multi-user.target +` + +const collectorWatchdogTimer = `[Unit] +Description=orca watchdog 30s cadence (P09, C-11) + +[Timer] +OnBootSec=30s +OnUnitActiveSec=30s +AccuracySec=5s +Unit=orca-watchdog.service + +[Install] +WantedBy=timers.target +` diff --git a/internal/cli/collector_test.go b/internal/cli/collector_test.go new file mode 100644 index 0000000..eb15e45 --- /dev/null +++ b/internal/cli/collector_test.go @@ -0,0 +1,160 @@ +package cli + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +func TestCollectorCmdRegistered(t *testing.T) { + found := false + for _, cmd := range rootCmd.Commands() { + if cmd.Name() == "collector" { + found = true + break + } + } + if !found { + t.Fatal("collector command not registered on root") + } +} + +func TestCollectorSubcommandsRegistered(t *testing.T) { + want := map[string]bool{"start": false, "stop": false, "status": false} + for _, cmd := range collectorCmd.Commands() { + if _, ok := want[cmd.Name()]; ok { + want[cmd.Name()] = true + } + } + for name, found := range want { + if !found { + t.Errorf("collector subcommand %q not registered", name) + } + } +} + +func runCollectorCmd(t *testing.T, root string, dryRun bool, args ...string) (string, error) { + t.Helper() + resetRootFlags(t) + full := append([]string{"collector"}, args...) + if root != "" { + full = append(full, "--root", root) + } + if dryRun && (len(args) > 0 && (args[0] == "start" || args[0] == "stop")) { + full = append(full, "--dry-run") + } + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs(full) + err := rootCmd.Execute() + return buf.String(), err +} + +func TestCollectorStartEmitsArtifacts(t *testing.T) { + root := t.TempDir() + out, err := runCollectorCmd(t, root, true, "start") + if err != nil { + t.Fatalf("orca collector start: %v\n%s", err, out) + } + wantFiles := []string{ + filepath.Join(root, collectorScriptDir, collectorAggregateSh), + filepath.Join(root, collectorScriptDir, collectorWatchdogSh), + filepath.Join(root, collectorUnitDir, collectorAggregateSvc), + filepath.Join(root, collectorUnitDir, collectorAggregateTmr), + filepath.Join(root, collectorUnitDir, collectorWatchdogSvc), + filepath.Join(root, collectorUnitDir, collectorWatchdogTmr), + } + for _, p := range wantFiles { + if _, err := os.Stat(p); err != nil { + t.Errorf("expected emitted file %s: %v", p, err) + } + } + for _, p := range []string{ + filepath.Join(root, collectorScriptDir, collectorAggregateSh), + filepath.Join(root, collectorScriptDir, collectorWatchdogSh), + } { + info, err := os.Stat(p) + if err != nil { + t.Fatalf("stat %s: %v", p, err) + } + if perm := info.Mode().Perm(); perm&0o111 == 0 { + t.Errorf("expected executable bit on %s, got %o", p, perm) + } + } +} + +func TestCollectorStartStatusRunning(t *testing.T) { + root := t.TempDir() + if _, err := runCollectorCmd(t, root, true, "start"); err != nil { + t.Fatalf("start: %v", err) + } + agg, wd := collectorRunning(root) + if !agg || !wd { + t.Errorf("expected both running, got agg=%t wd=%t", agg, wd) + } + out, err := runCollectorCmd(t, root, true, "status") + if err != nil { + t.Fatalf("status: %v", err) + } + if !contains(out, "running") { + t.Errorf("status output should say running, got %q", out) + } +} + +func TestCollectorStopRemovesArtifacts(t *testing.T) { + root := t.TempDir() + if _, err := runCollectorCmd(t, root, true, "start"); err != nil { + t.Fatalf("start: %v", err) + } + out, err := runCollectorCmd(t, root, true, "stop") + if err != nil { + t.Fatalf("stop: %v\n%s", err, out) + } + wantFiles := []string{ + filepath.Join(root, collectorScriptDir, collectorAggregateSh), + filepath.Join(root, collectorScriptDir, collectorWatchdogSh), + filepath.Join(root, collectorUnitDir, collectorAggregateSvc), + filepath.Join(root, collectorUnitDir, collectorAggregateTmr), + filepath.Join(root, collectorUnitDir, collectorWatchdogSvc), + filepath.Join(root, collectorUnitDir, collectorWatchdogTmr), + } + for _, p := range wantFiles { + if _, err := os.Stat(p); !os.IsNotExist(err) { + t.Errorf("expected %s removed, got %v", p, err) + } + } +} + +func TestCollectorStatusWhenStopped(t *testing.T) { + root := t.TempDir() + out, err := runCollectorCmd(t, root, true, "status") + if err != nil { + t.Fatalf("status: %v", err) + } + if !contains(out, "stopped") { + t.Errorf("expected stopped, got %q", out) + } + agg, wd := collectorRunning(root) + if agg || wd { + t.Errorf("expected neither running, got agg=%t wd=%t", agg, wd) + } +} + +func TestCollectorResolveRootRejectsRelative(t *testing.T) { + collectorRoot = "tmp/relative" + _, err := collectorResolveRoot() + if err == nil { + t.Error("expected error for relative root") + } + collectorRoot = "/" + r, err := collectorResolveRoot() + if err != nil || r != "/" { + t.Errorf("expected / for default, got %q err=%v", r, err) + } +} + +func contains(haystack, needle string) bool { + return bytes.Contains([]byte(haystack), []byte(needle)) +} diff --git a/internal/cli/namespace_test.go b/internal/cli/namespace_test.go index 2bed420..c24a552 100644 --- a/internal/cli/namespace_test.go +++ b/internal/cli/namespace_test.go @@ -43,6 +43,8 @@ func resetCommandFlags() { backupOutPath, restoreInPath, restoreTargetDir = "", "", "" restoreForce = false restoreDryRun = false + collectorRoot = "/" + collectorDryRun = false resetNSFlags() // Reset per-command output writers so tests that polluted them // (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into diff --git a/scripts/orca-aggregate.sh b/scripts/orca-aggregate.sh new file mode 100755 index 0000000..d1d86ee --- /dev/null +++ b/scripts/orca-aggregate.sh @@ -0,0 +1,172 @@ +#!/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/.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//manifest.json, +# triggers orca-remediate.sh 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" diff --git a/scripts/orca-watchdog.sh b/scripts/orca-watchdog.sh new file mode 100755 index 0000000..42c65a3 --- /dev/null +++ b/scripts/orca-watchdog.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +# orca-watchdog.sh — lead-side watchdog meta-timer (P09, C-11). +# +# Runs every 30s via orca-watchdog.timer. Checks whether +# orca-aggregate.sh has run in the last N seconds (default 30s — 3x the +# 10s cadence). If not: fires a structured alert via +# `logger -t orca-watchdog -p user.err "aggregator starvation detected"`. +# The check compares the mtime of /etc/orca/state/cluster.json to the +# current time. + +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/watchdog" + +STATE_DIR="${ORCA_STATE_DIR:-/etc/orca/state}" +CLUSTER_JSON="$STATE_DIR/cluster.json" +STARVATION_THRESHOLD="${ORCA_WATCHDOG_THRESHOLD:-30}" + +alert() { + logger -t orca-watchdog -p user.err "aggregator starvation detected: $1" +} + +if [ ! -f "$CLUSTER_JSON" ]; then + alert "cluster.json absent at $CLUSTER_JSON" + orca_log_error "watchdog" "-" "starved" "cluster.json absent" + exit 0 +fi + +now="$(date +%s)" +mtime="$(stat -c %Y "$CLUSTER_JSON" 2>/dev/null || stat -f %m "$CLUSTER_JSON" 2>/dev/null || echo 0)" +age=$((now - mtime)) + +if [ "$age" -ge "$STARVATION_THRESHOLD" ]; then + alert "cluster.json age ${age}s >= threshold ${STARVATION_THRESHOLD}s" + orca_log_error "watchdog" "-" "starved" "age=${age}s threshold=${STARVATION_THRESHOLD}s" +else + orca_log_info "watchdog" "-" "ok" "age=${age}s threshold=${STARVATION_THRESHOLD}s" +fi diff --git a/scripts/tests/aggregator_test.bash b/scripts/tests/aggregator_test.bash new file mode 100644 index 0000000..bf042c6 --- /dev/null +++ b/scripts/tests/aggregator_test.bash @@ -0,0 +1,126 @@ +#!/usr/bin/env bats +# Tests for scripts/orca-aggregate.sh + orca-watchdog.sh (P09, C-11, +# REQ-107). Hermetic: exercises the aggregator against a fake peers.env +# + state dir in a temp dir; the SSH path is exercised only when a +# fake ssh shim is on PATH. The watchdog is exercised with a stale +# cluster.json to confirm starvation detection. + +load test_helper + +@test "orca-aggregate.sh exists and is executable" { + [ -f "$SCRIPTS_DIR/orca-aggregate.sh" ] + [ -x "$SCRIPTS_DIR/orca-aggregate.sh" ] +} + +@test "orca-watchdog.sh exists and is executable" { + [ -f "$SCRIPTS_DIR/orca-watchdog.sh" ] + [ -x "$SCRIPTS_DIR/orca-watchdog.sh" ] +} + +@test "orca-aggregate.sh handles missing peers.env gracefully" { + tmp="$(mktemp -d)" + run env \ + ORCA_PEERS_ENV="$tmp/nonexistent.env" \ + ORCA_STATE_DIR="$tmp/state" \ + ORCA_REMOTE_STATE_DIR="$tmp/remote-state" \ + ORCA_REMOTE_DRIFT_DIR="$tmp/remote-drift" \ + ORCA_APPLIED_DIR="$tmp/applied" \ + bash "$SCRIPTS_DIR/orca-aggregate.sh" + assert_status 0 "$status" + # cluster.json written atomically with an empty peers array. + [ -f "$tmp/state/cluster.json" ] + assert_contains "$(cat "$tmp/state/cluster.json")" '"peers":[]' +} + +@test "orca-aggregate.sh skips drift section when drift-events absent" { + tmp="$(mktemp -d)" + # No applied dir and no remote drift dir on lead → drift section + # must skip (no drift-events-aggregated.json written). + run env \ + ORCA_PEERS_ENV="$tmp/nonexistent.env" \ + ORCA_STATE_DIR="$tmp/state" \ + ORCA_REMOTE_STATE_DIR="$tmp/remote-state" \ + ORCA_REMOTE_DRIFT_DIR="$tmp/absent-drift" \ + ORCA_APPLIED_DIR="$tmp/absent-applied" \ + bash "$SCRIPTS_DIR/orca-aggregate.sh" + assert_status 0 "$status" + [ -f "$tmp/state/cluster.json" ] + # Drift aggregation skipped → aggregated file must NOT exist. + [ ! -f "$tmp/state/drift-events-aggregated.json" ] +} + +@test "orca-watchdog.sh detects starvation with old cluster.json" { + tmp="$(mktemp -d)" + mkdir -p "$tmp/state" + cluster="$tmp/state/cluster.json" + echo '{}' >"$cluster" + # Set mtime to 120s ago so age (120) >= threshold (default 30). + old_ts="$(date -d '120 seconds ago' +%Y%m%d%H%M.%S)" + touch -t "$old_ts" "$cluster" + # logger is available in CI; capture the alert by overriding PATH + # with a fake logger that records to a file. + fake_bin="$tmp/bin" + mkdir -p "$fake_bin" + cat >"$fake_bin/logger" <<'LOGGER_EOF' +#!/usr/bin/env bash +echo "logger: $*" >>"$LOGGER_OUT" +LOGGER_EOF + chmod +x "$fake_bin/logger" + env \ + ORCA_STATE_DIR="$tmp/state" \ + ORCA_WATCHDOG_THRESHOLD=30 \ + LOGGER_OUT="$tmp/logger.out" \ + PATH="$fake_bin:$PATH" \ + bash "$SCRIPTS_DIR/orca-watchdog.sh" + [ -f "$tmp/logger.out" ] + out="$(cat "$tmp/logger.out")" + assert_contains "$out" "aggregator starvation detected" +} + +@test "orca-watchdog.sh passes when cluster.json is fresh" { + tmp="$(mktemp -d)" + mkdir -p "$tmp/state" + echo '{}' >"$tmp/state/cluster.json" + # Fresh mtime (now); age 0 < threshold. + fake_bin="$tmp/bin" + mkdir -p "$fake_bin" + cat >"$fake_bin/logger" <<'LOGGER_EOF' +#!/usr/bin/env bash +echo "logger: $*" >>"$LOGGER_OUT" +LOGGER_EOF + chmod +x "$fake_bin/logger" + env \ + ORCA_STATE_DIR="$tmp/state" \ + ORCA_WATCHDOG_THRESHOLD=30 \ + LOGGER_OUT="$tmp/logger.out" \ + PATH="$fake_bin:$PATH" \ + bash "$SCRIPTS_DIR/orca-watchdog.sh" + if [ -f "$tmp/logger.out" ]; then + out="$(cat "$tmp/logger.out")" + ! assert_contains "$out" "aggregator starvation detected" || { + echo "unexpected starvation alert for fresh cluster.json" >&2 + return 1 + } + fi +} + +@test "orca-watchdog.sh alerts when cluster.json absent" { + tmp="$(mktemp -d)" + fake_bin="$tmp/bin" + mkdir -p "$fake_bin" + cat >"$fake_bin/logger" <<'LOGGER_EOF' +#!/usr/bin/env bash +echo "logger: $*" >>"$LOGGER_OUT" +LOGGER_EOF + chmod +x "$fake_bin/logger" + env \ + ORCA_STATE_DIR="$tmp/state" \ + ORCA_WATCHDOG_THRESHOLD=30 \ + LOGGER_OUT="$tmp/logger.out" \ + PATH="$fake_bin:$PATH" \ + bash "$SCRIPTS_DIR/orca-watchdog.sh" + [ -f "$tmp/logger.out" ] + out="$(cat "$tmp/logger.out")" + assert_contains "$out" "aggregator starvation detected" + assert_contains "$out" "cluster.json absent" +}