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---
This commit is contained in:
Jon Chery
@@ -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
|
||||
// <root> 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
|
||||
`
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Executable
+172
@@ -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/<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"
|
||||
Executable
+42
@@ -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
|
||||
@@ -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"
|
||||
}
|
||||
Reference in New Issue
Block a user