orca/internal/cli/collector.go

// 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
`