// Package cli: drift.go implements the `orca drift` subcommand family // (P10b, v0.11; R-018/R-019/R-020, REQ-104). Subcommands: // // orca drift show current drift state (table) // orca drift watch [--interval=2s] [--paths=...] [--json] // stream drift events (iter.Seq2, D-017) // ctrl-c cancels via signal.NotifyContext (D-023) // orca drift show [--peer ] detailed drift events for a peer // orca drift acknowledge // record operator acknowledgment // orca drift remediate [--force] // trigger manual remediation // orca drift config show show current drift config // orca drift config validate validate config // // Plus the `orca job restart ` command for EnvironmentFile drift // (REQ-113, D-235) — restarts an allocation to pick up env-file drift. package cli import ( "context" "encoding/json" "errors" "fmt" "os" "os/signal" "path/filepath" "strings" "syscall" "time" "github.com/spf13/cobra" "git.cloudinit.dev/coreci/orca/internal/certpaths" "git.cloudinit.dev/coreci/orca/internal/drift" "git.cloudinit.dev/coreci/orca/internal/sshpush" ) var ( driftWatchInterval time.Duration driftWatchPaths []string driftShowPeer string driftConfigPath string driftRemediateForce bool driftAckPeer string driftAckPath string driftRemediatePeer string driftRemediatePath string driftWatchPollOverride time.Duration ) // driftTransport is the SSH-push surface the drift CLI needs. It // mirrors drift.Transport; tests substitute a mock. type driftTransport interface { Exec(ctx context.Context, peer string, cmd string) ([]byte, error) WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) ReadFile(ctx context.Context, peer string, path string) ([]byte, error) } // driftTransportOverride is the package-level test seam. var driftTransportOverride driftTransport func driftTransportFromCtx() (driftTransport, error) { if driftTransportOverride != nil { return driftTransportOverride, nil } keyPath := certpaths.SSHKeyPath() khPath := certpaths.KnownHostsPath() return sshpush.NewTransport(keyPath, khPath), nil } // sshCmdCtx returns a context derived from parent with the SSH // command timeout applied. If d <= 0, the parent is returned unchanged // (no deadline). REQ-157 / P08 T6: gives SSH-driven CLI subcommands a // bounded deadline so a hung peer cannot block forever. func sshCmdCtx(parent context.Context, d time.Duration) (context.Context, context.CancelFunc) { if d <= 0 { return context.WithCancel(parent) } return context.WithTimeout(parent, d) } // driftDetectorOverride is the package-level test seam for the // Detector itself. When non-nil it replaces the production detector // (which wraps a driftTransport). Tests set it and restore nil. var driftDetectorOverride drift.Detector func driftDetector() (drift.Detector, error) { if driftDetectorOverride != nil { return driftDetectorOverride, nil } t, err := driftTransportFromCtx() if err != nil { return nil, err } return drift.NewDefaultDetector(t), nil } var driftCmd = &cobra.Command{ Use: "drift", Short: "Detect and remediate control-plane drift (P10b)", Long: `Orca's drift detector is a BACKSTOP (R-019): the primary consistency mechanism is systemd / Traefik / step-ca / Syncthing themselves. The detector polls the lead's aggregated drift state (drift-events-aggregated.json) and can trigger orca-remediate.sh for auto-remediable paths. Pre-flight drift blocks txn apply (R-020).`, } var driftShowCmd = &cobra.Command{ Use: "show", Short: "Show current drift state (table format)", Long: `Show the aggregated drift events from the lead. Use --peer to filter to a single peer.`, Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, args []string) error { d, err := driftDetector() if err != nil { return fmt.Errorf("drift detector: %w", err) } events, err := d.Aggregate(cmd.Context(), driftShowPeer) if err != nil { return fmt.Errorf("aggregate: %w", err) } if driftShowPeer != "" { var filtered []drift.Event for _, e := range events { if e.Host == driftShowPeer { filtered = append(filtered, e) } } events = filtered } if jsonOutput { return printJSON(events) } out := cmd.OutOrStdout() if len(events) == 0 { fmt.Fprintln(out, "No drift events.") return nil } fmt.Fprintf(out, "%-20s %-20s %-40s %-10s %-10s\n", "EVENT-ID", "HOST", "PATH", "STATUS", "CONFIRMED") for _, e := range events { path := e.Path if len(path) > 40 { path = "..." + path[len(path)-37:] } confirmed := "no" if e.DriftConfirmed { confirmed = "yes" } fmt.Fprintf(out, "%-20s %-20s %-40s %-10s %-10s\n", e.EventID, e.Host, path, e.Status, confirmed) } return nil }, } var driftWatchCmd = &cobra.Command{ Use: "watch", Short: "Stream drift events (ctrl-c to cancel)", Long: `Stream drift events from the lead's aggregated state. Default poll is 2s; override with --interval. Use --paths=, to filter. Uses iter.Seq2 (D-017) and signal.NotifyContext for ctrl-c (D-023).`, Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, args []string) error { d, err := driftDetector() if err != nil { return fmt.Errorf("drift detector: %w", err) } ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) defer cancel() interval := driftWatchInterval if interval <= 0 { interval = 2 * time.Second } var specs []drift.PathSpec for _, p := range driftWatchPaths { if p == "" { continue } specs = append(specs, drift.PathSpec{Pattern: p, Interval: interval}) } out := cmd.OutOrStdout() for e, err := range d.Watch(ctx, specs) { if err != nil { if errors.Is(err, context.Canceled) { return nil } fmt.Fprintf(out, "watch error: %v\n", err) continue } if jsonOutput { line, _ := json.Marshal(e) fmt.Fprintln(out, string(line)) } else { fmt.Fprintf(out, "%s [%s] %s %s %s confirmed=%t\n", e.TS.Format(time.RFC3339), e.EventID, e.Host, e.Path, e.Status, e.DriftConfirmed) } } return nil }, } var driftAckCmd = &cobra.Command{ Use: "acknowledge ", Short: "Record operator acknowledgment of drift on a peer", Long: `Record operator acknowledgment for the given path in drift-acknowledgments.json on the lead. Acknowledged drift no longer blocks txn apply for that namespace (R-020).`, Args: cobra.ExactArgs(2), RunE: func(cmd *cobra.Command, args []string) error { peer := args[0] path := args[1] d, err := driftDetector() if err != nil { return fmt.Errorf("drift detector: %w", err) } ctx, cancel := sshCmdCtx(cmd.Context(), driftAckTimeout) defer cancel() if err := d.Acknowledge(ctx, peer, path); err != nil { return fmt.Errorf("acknowledge: %w", err) } printResult(fmt.Sprintf("✓ Acknowledged drift on %s for %s", peer, path), map[string]any{ "peer": peer, "path": path, "status": "acknowledged", }) return nil }, } var driftRemediateCmd = &cobra.Command{ Use: "remediate ", Short: "Trigger manual remediation of drift on a peer", Long: `Trigger orca-remediate.sh on the lead for the given path. --force bypasses the cooldown window (C4).`, Args: cobra.ExactArgs(2), RunE: func(cmd *cobra.Command, args []string) error { peer := args[0] path := args[1] d, err := driftDetector() if err != nil { return fmt.Errorf("drift detector: %w", err) } ctx, cancel := sshCmdCtx(cmd.Context(), driftRemediateTimeout) defer cancel() if err := d.Remediate(ctx, peer, path, driftRemediateForce); err != nil { if errors.Is(err, drift.ErrCooldown) { printResult(fmt.Sprintf("✗ Remediation in cooldown for %s on %s (use --force to bypass)", path, peer), map[string]any{ "peer": peer, "path": path, "status": "cooldown", }) return nil } return fmt.Errorf("remediate: %w", err) } printResult(fmt.Sprintf("✓ Remediated drift on %s for %s", peer, path), map[string]any{ "peer": peer, "path": path, "status": "remediated", "force": driftRemediateForce, }) return nil }, } var driftConfigCmd = &cobra.Command{ Use: "config", Short: "Show or validate the drift config", } var driftConfigShowCmd = &cobra.Command{ Use: "show", Short: "Show the current drift config", Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, args []string) error { cfg, err := drift.LoadConfig(driftConfigPath) if err != nil { return fmt.Errorf("load config: %w", err) } if jsonOutput { return printJSON(cfg) } out := cmd.OutOrStdout() fmt.Fprintf(out, "Polling: enabled=%t default=%s max_peers=%d\n", cfg.Polling.Enabled, cfg.Polling.DefaultInterval, cfg.Polling.MaxConcurrentPeers) fmt.Fprintln(out, "Critical paths:") for _, p := range cfg.Paths.Critical { fmt.Fprintf(out, " [%s] %s (interval=%s, path_unit=%t)\n", p.Tier, p.Pattern, p.Interval, p.SystemdPathUnit) } fmt.Fprintln(out, "Standard paths:") for _, p := range cfg.Paths.Standard { fmt.Fprintf(out, " [%s] %s (interval=%s)\n", p.Tier, p.Pattern, p.Interval) } fmt.Fprintln(out, "Excluded paths:") for _, p := range cfg.Paths.Excluded { fmt.Fprintf(out, " %s\n", p) } fmt.Fprintf(out, "Remediation: auto=%t notify=%t\n", cfg.Remediate.Auto, cfg.Remediate.NotifyOnRemediation) if len(cfg.Remediate.AutoPaths) > 0 { fmt.Fprintln(out, " auto_paths:") for _, p := range cfg.Remediate.AutoPaths { fmt.Fprintf(out, " %s\n", p) } } if len(cfg.Remediate.RequireApproval) > 0 { fmt.Fprintln(out, " require_approval:") for _, p := range cfg.Remediate.RequireApproval { fmt.Fprintf(out, " %s\n", p) } } return nil }, } var driftConfigValidateCmd = &cobra.Command{ Use: "validate", Short: "Validate the drift config", Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, args []string) error { cfg, err := drift.LoadConfig(driftConfigPath) if err != nil { return fmt.Errorf("load config: %w", err) } if err := drift.ValidateConfig(cfg); err != nil { printResult(fmt.Sprintf("✗ Config invalid: %v", err), map[string]any{"valid": false, "error": err.Error()}) return err } printResult("✓ Config valid", map[string]any{"valid": true}) return nil }, } // jobRestartCmd implements `orca job restart ` (REQ-113, D-235): // restart an allocation on its peer to pick up EnvironmentFile drift. var jobRestartCmd = &cobra.Command{ Use: "restart ", Short: "Restart an allocation to pick up EnvironmentFile drift (REQ-113)", Long: `SSH to the peer running allocation and run systemctl restart orca-alloc-.service. This is the normal allocation lifecycle (NOT file-level remediation) and is triggered when /etc/orca/allocs//env drifts.`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { name := args[0] peer := jobRestartPeer if peer == "" { return fmt.Errorf("--peer is required for job restart") } transport, err := driftTransportFromCtx() if err != nil { return fmt.Errorf("ssh transport: %w", err) } unit := fmt.Sprintf("orca-alloc-%s.service", name) restartCmd := fmt.Sprintf("systemctl restart %s", shellQuoteDrift(unit)) ctx, cancel := sshCmdCtx(cmd.Context(), jobRestartTimeout) defer cancel() out, err := transport.Exec(ctx, peer, restartCmd) if err != nil { return fmt.Errorf("restart %s on %s: %w (output: %s)", unit, peer, err, string(out)) } printResult(fmt.Sprintf("✓ Restarted %s on %s", unit, peer), map[string]any{ "unit": unit, "peer": peer, "status": "restarted", "output": string(out), }) return nil }, } var jobRestartPeer string var driftRemediateTimeout time.Duration var driftAckTimeout time.Duration var jobRestartTimeout time.Duration func shellQuoteDrift(s string) string { return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" } func init() { driftWatchCmd.Flags().DurationVar(&driftWatchInterval, "interval", 2*time.Second, "poll interval (default 2s)") driftWatchCmd.Flags().StringSliceVar(&driftWatchPaths, "paths", nil, "comma-separated glob patterns to watch (default: all)") driftShowCmd.Flags().StringVar(&driftShowPeer, "peer", "", "filter to a single peer host") driftRemediateCmd.Flags().BoolVar(&driftRemediateForce, "force", false, "bypass the cooldown window (C4)") driftRemediateCmd.Flags().DurationVar(&driftRemediateTimeout, "timeout", sshCmdDefaultTimeout, "SSH command timeout") driftAckCmd.Flags().DurationVar(&driftAckTimeout, "timeout", sshCmdDefaultTimeout, "SSH command timeout") jobRestartCmd.Flags().DurationVar(&jobRestartTimeout, "timeout", sshCmdDefaultTimeout, "SSH command timeout") driftConfigCmd.PersistentFlags().StringVar(&driftConfigPath, "config", "", "path to drift config JSON (default: built-in)") jobRestartCmd.Flags().StringVar(&jobRestartPeer, "peer", "", "peer address (host:port) running the allocation") driftCmd.AddCommand(driftShowCmd) driftCmd.AddCommand(driftWatchCmd) driftCmd.AddCommand(driftAckCmd) driftCmd.AddCommand(driftRemediateCmd) driftCmd.AddCommand(driftConfigCmd) driftConfigCmd.AddCommand(driftConfigShowCmd) driftConfigCmd.AddCommand(driftConfigValidateCmd) rootCmd.AddCommand(driftCmd) jobCmd.AddCommand(jobRestartCmd) } // writeClusterDefaultDriftConfig writes the canonical default drift // config to the cluster state dir so the lead has a reference copy. // Best-effort; caller logs failures. func writeClusterDefaultDriftConfig() error { dir := filepath.Join(os.Getenv("ORCA_HOME"), "cluster", "state") if err := os.MkdirAll(dir, 0o755); err != nil { return fmt.Errorf("mkdir cluster state: %w", err) } path := filepath.Join(dir, "drift.json") cfg := drift.DefaultConfig() raw, err := json.MarshalIndent(cfg, "", " ") if err != nil { return fmt.Errorf("marshal drift config: %w", err) } tmp := path + ".tmp" if err := os.WriteFile(tmp, raw, 0o644); err != nil { return fmt.Errorf("write drift config: %w", err) } if err := os.Rename(tmp, path); err != nil { return fmt.Errorf("rename drift config: %w", err) } return nil }