feat(P14b,P14c): daemon cutover + rotate-lead (REQ-114) + mixed-version tolerance (REQ-065, REQ-086, C-13)
P14b: orca cluster cutover (stop v0.8 daemon, adopt running allocs); orca cluster rotate-lead --to (R-003 enforcement, CA+master key copy, SSH key rotation). P14c: orca doctor no-orca-on-server (R-001 enforcement); orca cluster compat-check (mixed-version tolerance). ---ci--- project: orca phase: 14b milestone: v0.11 status: execute ---/ci---
This commit is contained in:
Jon Chery
@@ -0,0 +1,17 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
var clusterCmd = &cobra.Command{
|
||||
Use: "cluster",
|
||||
Short: "Cluster-wide operations (cutover, rotate-lead, compat-check)",
|
||||
Long: `Cluster-wide operations: daemon cutover, lead rotation, and
|
||||
mixed-version compatibility checks.`,
|
||||
}
|
||||
|
||||
func init() {
|
||||
clusterCmd.AddCommand(clusterCutoverCmd, clusterRotateLeadCmd, compatCheckCmd)
|
||||
rootCmd.AddCommand(clusterCmd)
|
||||
}
|
||||
@@ -0,0 +1,417 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/emit"
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
)
|
||||
|
||||
var noOrcaOnServerCmd = &cobra.Command{
|
||||
Use: "no-orca-on-server",
|
||||
Short: "Verify no orca binary/service/process on peers (REQ-086, R-001, C-13)",
|
||||
Long: `SSH to each registered peer and verify that no orca binary,
|
||||
systemd service, or process is present on the server (R-001: no orca
|
||||
binary on any server; C-13 enforcement).
|
||||
|
||||
Checks per peer:
|
||||
1. command -v orca → must return nothing (no orca in PATH)
|
||||
2. systemctl list-units 'orca*' (excluding orca-alloc-*) → must be empty
|
||||
3. pgrep orca → must return nothing (no orca process)
|
||||
4. /etc/orca/ contains no orca binaries (config dir is OK)
|
||||
|
||||
A peer with any violation is reported as FAIL. The exit code is non-zero
|
||||
if any peer fails.`,
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return runNoOrcaOnServer(cmd)
|
||||
},
|
||||
}
|
||||
|
||||
type noOrcaPeerResult struct {
|
||||
Node string `json:"node"`
|
||||
Peer string `json:"peer"`
|
||||
Pass bool `json:"pass"`
|
||||
Violations []string `json:"violations,omitempty"`
|
||||
}
|
||||
|
||||
func runNoOrcaOnServer(cmd *cobra.Command) error {
|
||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
reg, closer, err := nodeRegistry()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer closer()
|
||||
|
||||
nodes, err := reg.List(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list nodes: %w", err)
|
||||
}
|
||||
|
||||
ex, err := drainExecFromCtx(cmd.Context())
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh transport: %w", err)
|
||||
}
|
||||
|
||||
log := newLogger()
|
||||
results := make([]noOrcaPeerResult, 0, len(nodes))
|
||||
var failedNodes []string
|
||||
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
peer := peerAddrForNode(n)
|
||||
if peer == "" {
|
||||
continue
|
||||
}
|
||||
r := noOrcaPeerResult{Node: n.Name, Peer: peer, Pass: true, Violations: []string{}}
|
||||
|
||||
if v, ok := checkNoOrcaBinary(ctx, ex, peer); !ok {
|
||||
r.Pass = false
|
||||
r.Violations = append(r.Violations, v)
|
||||
}
|
||||
if v, ok := checkNoOrcaService(ctx, ex, peer); !ok {
|
||||
r.Pass = false
|
||||
r.Violations = append(r.Violations, v)
|
||||
}
|
||||
if v, ok := checkNoOrcaProcess(ctx, ex, peer); !ok {
|
||||
r.Pass = false
|
||||
r.Violations = append(r.Violations, v)
|
||||
}
|
||||
if v, ok := checkNoOrcaBinInEtc(ctx, ex, peer); !ok {
|
||||
r.Pass = false
|
||||
r.Violations = append(r.Violations, v)
|
||||
}
|
||||
|
||||
if !r.Pass {
|
||||
failedNodes = append(failedNodes, n.Name)
|
||||
log.Warn("no-orca-on-server: violations",
|
||||
slog.String("node", n.Name), slog.Any("violations", r.Violations))
|
||||
}
|
||||
results = append(results, r)
|
||||
}
|
||||
|
||||
summary := map[string]any{
|
||||
"results": results,
|
||||
"failed": failedNodes,
|
||||
}
|
||||
|
||||
if jsonOutput {
|
||||
return printJSON(summary)
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
for _, r := range results {
|
||||
status := "PASS"
|
||||
if !r.Pass {
|
||||
status = "FAIL"
|
||||
}
|
||||
fmt.Fprintf(out, "%-20s %-5s %s\n", r.Node, status, strings.Join(r.Violations, "; "))
|
||||
}
|
||||
if len(failedNodes) > 0 {
|
||||
fmt.Fprintf(out, "\n%d peer(s) failed R-001 enforcement\n", len(failedNodes))
|
||||
return fmt.Errorf("no-orca-on-server: %d peer(s) have violations", len(failedNodes))
|
||||
}
|
||||
fmt.Fprintf(out, "\n✓ all peers clean (R-001 enforced)\n")
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkNoOrcaBinary(ctx context.Context, ex drainExecer, peer string) (string, bool) {
|
||||
out, err := ex.Exec(ctx, peer, "command -v orca 2>/dev/null || true")
|
||||
if err != nil {
|
||||
return "", true
|
||||
}
|
||||
if strings.TrimSpace(string(out)) != "" {
|
||||
return fmt.Sprintf("orca binary in PATH: %s", strings.TrimSpace(string(out))), false
|
||||
}
|
||||
return "", true
|
||||
}
|
||||
|
||||
func checkNoOrcaService(ctx context.Context, ex drainExecer, peer string) (string, bool) {
|
||||
cmd := "systemctl list-units 'orca*' --no-legend --no-pager 2>/dev/null | grep -v 'orca-alloc-' || true"
|
||||
out, err := ex.Exec(ctx, peer, cmd)
|
||||
if err != nil {
|
||||
return "", true
|
||||
}
|
||||
trimmed := strings.TrimSpace(string(out))
|
||||
if trimmed != "" {
|
||||
return fmt.Sprintf("orca systemd service(s) present: %s", trimmed), false
|
||||
}
|
||||
return "", true
|
||||
}
|
||||
|
||||
func checkNoOrcaProcess(ctx context.Context, ex drainExecer, peer string) (string, bool) {
|
||||
out, err := ex.Exec(ctx, peer, "pgrep -x orca 2>/dev/null || true")
|
||||
if err != nil {
|
||||
return "", true
|
||||
}
|
||||
if strings.TrimSpace(string(out)) != "" {
|
||||
return fmt.Sprintf("orca process running: pid(s) %s", strings.TrimSpace(string(out))), false
|
||||
}
|
||||
return "", true
|
||||
}
|
||||
|
||||
func checkNoOrcaBinInEtc(ctx context.Context, ex drainExecer, peer string) (string, bool) {
|
||||
cmd := "find /etc/orca -type f -executable 2>/dev/null | grep -v 'scripts/' | head -5 || true"
|
||||
out, err := ex.Exec(ctx, peer, cmd)
|
||||
if err != nil {
|
||||
return "", true
|
||||
}
|
||||
trimmed := strings.TrimSpace(string(out))
|
||||
if trimmed != "" {
|
||||
return fmt.Sprintf("executable(s) under /etc/orca: %s", trimmed), false
|
||||
}
|
||||
return "", true
|
||||
}
|
||||
|
||||
var compatCheckCmd = &cobra.Command{
|
||||
Use: "compat-check",
|
||||
Short: "Check mixed-version tolerance across peers (REQ-065, C-13)",
|
||||
Long: `Check that the cluster tolerates mixed orca versions during an
|
||||
upgrade window (REQ-065). The lead and peers may run different orca
|
||||
versions during a rolling upgrade; this command verifies:
|
||||
|
||||
- Each peer's orca version (reported)
|
||||
- The txn manifest format is compatible across versions
|
||||
- The render-contract JSON schema (emit.SchemaVersion) is versioned
|
||||
and backward-compatible
|
||||
- No new required fields that old peers don't understand
|
||||
|
||||
Reports: which peers are on which version, any compatibility issues.`,
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return runCompatCheck(cmd)
|
||||
},
|
||||
}
|
||||
|
||||
type compatPeerResult struct {
|
||||
Node string `json:"node"`
|
||||
Peer string `json:"peer"`
|
||||
Version string `json:"version"`
|
||||
LeadVersion string `json:"lead_version,omitempty"`
|
||||
Compatible bool `json:"compatible"`
|
||||
Issue string `json:"issue,omitempty"`
|
||||
}
|
||||
|
||||
func runCompatCheck(cmd *cobra.Command) error {
|
||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
reg, closer, err := nodeRegistry()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer closer()
|
||||
|
||||
nodes, err := reg.List(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list nodes: %w", err)
|
||||
}
|
||||
|
||||
ex, err := drainExecFromCtx(cmd.Context())
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh transport: %w", err)
|
||||
}
|
||||
|
||||
leadVersion := version
|
||||
results := make([]compatPeerResult, 0, len(nodes))
|
||||
var issues []string
|
||||
versionSet := map[string]int{}
|
||||
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
peer := peerAddrForNode(n)
|
||||
if peer == "" {
|
||||
continue
|
||||
}
|
||||
peerVersion := detectPeerOrcaVersion(ctx, ex, peer)
|
||||
versionSet[peerVersion]++
|
||||
r := compatPeerResult{
|
||||
Node: n.Name,
|
||||
Peer: peer,
|
||||
Version: peerVersion,
|
||||
LeadVersion: leadVersion,
|
||||
Compatible: true,
|
||||
}
|
||||
if peerVersion != "" && peerVersion != leadVersion {
|
||||
if !versionsCompatible(leadVersion, peerVersion) {
|
||||
r.Compatible = false
|
||||
r.Issue = fmt.Sprintf("peer %s (%s) incompatible with lead (%s)",
|
||||
n.Name, peerVersion, leadVersion)
|
||||
issues = append(issues, r.Issue)
|
||||
}
|
||||
}
|
||||
results = append(results, r)
|
||||
}
|
||||
|
||||
schemaOK := verifyRenderContractCompat(ctx, ex, nodes)
|
||||
if !schemaOK {
|
||||
issues = append(issues, "render-contract schema mismatch detected across peers")
|
||||
}
|
||||
manifestOK := verifyTxnManifestCompat(ctx, ex, nodes)
|
||||
if !manifestOK {
|
||||
issues = append(issues, "txn manifest format incompatibility detected")
|
||||
}
|
||||
|
||||
summary := map[string]any{
|
||||
"lead_version": leadVersion,
|
||||
"schema_version": emit.SchemaVersion,
|
||||
"results": results,
|
||||
"versions_seen": versionSet,
|
||||
"issues": issues,
|
||||
"schema_ok": schemaOK,
|
||||
"manifest_ok": manifestOK,
|
||||
}
|
||||
|
||||
if jsonOutput {
|
||||
return printJSON(summary)
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
fmt.Fprintf(out, "lead version: %s (schema %s)\n", leadVersion, emit.SchemaVersion)
|
||||
for _, r := range results {
|
||||
mark := "✓"
|
||||
if !r.Compatible {
|
||||
mark = "✗"
|
||||
}
|
||||
fmt.Fprintf(out, " %s %-20s %s\n", mark, r.Node, r.Version)
|
||||
if r.Issue != "" {
|
||||
fmt.Fprintf(out, " %s\n", r.Issue)
|
||||
}
|
||||
}
|
||||
if len(issues) > 0 {
|
||||
fmt.Fprintf(out, "\n%d compatibility issue(s) found\n", len(issues))
|
||||
return fmt.Errorf("compat-check: %d issue(s)", len(issues))
|
||||
}
|
||||
fmt.Fprintf(out, "\n✓ all peers compatible\n")
|
||||
return nil
|
||||
}
|
||||
|
||||
func detectPeerOrcaVersion(ctx context.Context, ex drainExecer, peer string) string {
|
||||
out, err := ex.Exec(ctx, peer, "orca version --json 2>/dev/null || true")
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
s := strings.TrimSpace(string(out))
|
||||
if s == "" {
|
||||
return ""
|
||||
}
|
||||
var parsed map[string]any
|
||||
if err := json.Unmarshal([]byte(s), &parsed); err == nil {
|
||||
if v, ok := parsed["version"]; ok {
|
||||
if vs, ok := v.(string); ok && vs != "" {
|
||||
return vs
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, line := range strings.Split(s, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if strings.Contains(line, "version") {
|
||||
fields := strings.Fields(line)
|
||||
for i, f := range fields {
|
||||
if f == "\"version\":" || f == "version:" {
|
||||
if i+1 < len(fields) {
|
||||
return strings.Trim(fields[i+1], "\",")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func versionsCompatible(lead, peer string) bool {
|
||||
if lead == "" || peer == "" {
|
||||
return true
|
||||
}
|
||||
li := versionMinor(lead)
|
||||
pi := versionMinor(peer)
|
||||
if li == 0 || pi == 0 {
|
||||
return true
|
||||
}
|
||||
diff := li - pi
|
||||
if diff < 0 {
|
||||
diff = -diff
|
||||
}
|
||||
return diff <= 1
|
||||
}
|
||||
|
||||
func versionMinor(v string) int {
|
||||
s := strings.TrimPrefix(v, "v")
|
||||
parts := strings.Split(s, ".")
|
||||
if len(parts) < 2 {
|
||||
return 0
|
||||
}
|
||||
var n int
|
||||
for _, c := range parts[1] {
|
||||
if c >= '0' && c <= '9' {
|
||||
n = n*10 + int(c-'0')
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func verifyRenderContractCompat(ctx context.Context, ex drainExecer, nodes []*model.Node) bool {
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
peer := peerAddrForNode(n)
|
||||
if peer == "" {
|
||||
continue
|
||||
}
|
||||
out, err := ex.Exec(ctx, peer, "test -f /etc/orca/cluster/render-contract.json && cat /etc/orca/cluster/render-contract.json || true")
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
s := strings.TrimSpace(string(out))
|
||||
if s == "" {
|
||||
continue
|
||||
}
|
||||
if !strings.Contains(s, emit.SchemaVersion) && !strings.Contains(s, "schema_version") {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func verifyTxnManifestCompat(ctx context.Context, ex drainExecer, nodes []*model.Node) bool {
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
peer := peerAddrForNode(n)
|
||||
if peer == "" {
|
||||
continue
|
||||
}
|
||||
out, err := ex.Exec(ctx, peer, "test -d /etc/orca/cluster/txns && ls /etc/orca/cluster/txns | head -1 || true")
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
first := strings.TrimSpace(string(out))
|
||||
if first == "" {
|
||||
continue
|
||||
}
|
||||
man, err := ex.Exec(ctx, peer, fmt.Sprintf("cat /etc/orca/cluster/txns/%s/manifest.json 2>/dev/null || true", first))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
s := strings.TrimSpace(string(man))
|
||||
if s == "" {
|
||||
continue
|
||||
}
|
||||
if !strings.Contains(s, "txn_id") || !strings.Contains(s, "files") {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func sshQuote(s string) string {
|
||||
return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
|
||||
}
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/engine"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
var cutoverTimeout time.Duration
|
||||
|
||||
var clusterCutoverCmd = &cobra.Command{
|
||||
Use: "cutover",
|
||||
Short: "Stop v0.8 orca daemons and adopt running allocs (P14b)",
|
||||
Long: `Stop the v0.8 orca-daemon on every peer that still runs one,
|
||||
discover its running allocations (orca-alloc-*.service), and adopt each
|
||||
into the SSH-push path (mark it managed by the CLI-side scheduler).
|
||||
|
||||
The allocation's systemd unit keeps running independently of the
|
||||
daemon; the cutover only re-records ownership in the cluster store
|
||||
and stops the daemon.
|
||||
|
||||
Idempotent: a peer whose daemon is already stopped is a no-op for that
|
||||
peer. Re-adopting an already-adopted alloc is a no-op.`,
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return runCutover(cmd)
|
||||
},
|
||||
}
|
||||
|
||||
func runCutover(cmd *cobra.Command) error {
|
||||
ctx, cancel := context.WithTimeout(cmd.Context(), cutoverTimeout)
|
||||
defer cancel()
|
||||
|
||||
reg, closer, err := nodeRegistry()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer closer()
|
||||
|
||||
nodes, err := reg.List(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list nodes: %w", err)
|
||||
}
|
||||
|
||||
ex, err := drainExecFromCtx(cmd.Context())
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh transport: %w", err)
|
||||
}
|
||||
|
||||
log := newLogger()
|
||||
|
||||
type peerResult struct {
|
||||
Node string `json:"node"`
|
||||
Peer string `json:"peer"`
|
||||
DaemonStopped bool `json:"daemon_stopped"`
|
||||
AlreadyStopped bool `json:"already_stopped"`
|
||||
Adopted []string `json:"adopted"`
|
||||
Failed string `json:"failed,omitempty"`
|
||||
}
|
||||
|
||||
results := make([]peerResult, 0, len(nodes))
|
||||
var stopped, already, failed, adopted []string
|
||||
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
peer := peerAddrForNode(n)
|
||||
if peer == "" {
|
||||
continue
|
||||
}
|
||||
pr := peerResult{Node: n.Name, Peer: peer, Adopted: []string{}}
|
||||
|
||||
stopCmd := "systemctl stop orca-daemon.service"
|
||||
_, stopErr := ex.Exec(ctx, peer, stopCmd)
|
||||
switch {
|
||||
case stopErr == nil:
|
||||
pr.DaemonStopped = true
|
||||
stopped = append(stopped, n.Name)
|
||||
default:
|
||||
var exitErr *sshExitErr
|
||||
if errors.As(stopErr, &exitErr) && exitErr.code == 5 {
|
||||
pr.AlreadyStopped = true
|
||||
already = append(already, n.Name)
|
||||
} else {
|
||||
pr.Failed = stopErr.Error()
|
||||
failed = append(failed, n.Name)
|
||||
results = append(results, pr)
|
||||
log.Warn("cutover: stop daemon failed",
|
||||
slog.String("node", n.Name), slog.String("peer", peer), "error", stopErr)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
ids, listErr := listRunningAllocs(ctx, ex, peer)
|
||||
if listErr != nil {
|
||||
pr.Failed = listErr.Error()
|
||||
failed = append(failed, n.Name)
|
||||
results = append(results, pr)
|
||||
log.Warn("cutover: list allocs failed",
|
||||
slog.String("node", n.Name), slog.String("peer", peer), "error", listErr)
|
||||
continue
|
||||
}
|
||||
for _, id := range ids {
|
||||
if err := adoptAlloc(ctx, n.ID, id); err != nil {
|
||||
log.Warn("cutover: adopt alloc failed",
|
||||
slog.String("alloc", id), slog.String("node", n.Name), "error", err)
|
||||
pr.Failed = fmt.Sprintf("%sadopt %s: %v", pr.Failed, id, err)
|
||||
continue
|
||||
}
|
||||
pr.Adopted = append(pr.Adopted, id)
|
||||
adopted = append(adopted, n.Name+"/"+id)
|
||||
}
|
||||
results = append(results, pr)
|
||||
}
|
||||
|
||||
summary := map[string]any{
|
||||
"stopped": stopped,
|
||||
"already_stopped": already,
|
||||
"failed": failed,
|
||||
"adopted": adopted,
|
||||
"per_node": results,
|
||||
}
|
||||
|
||||
if db, dbErr := store.Open(certpaths.DBPath()); dbErr == nil {
|
||||
engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, "cli", "cluster.cutover", "cluster", "success", nil, summary)
|
||||
db.Close()
|
||||
}
|
||||
|
||||
if jsonOutput {
|
||||
return printJSON(summary)
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
fmt.Fprintf(out, "✓ cutover complete (%d stopped, %d already stopped, %d failed, %d adopted)\n",
|
||||
len(stopped), len(already), len(failed), len(adopted))
|
||||
for _, n := range stopped {
|
||||
fmt.Fprintf(out, " stopped %s\n", n)
|
||||
}
|
||||
for _, n := range already {
|
||||
fmt.Fprintf(out, " already-stopped %s\n", n)
|
||||
}
|
||||
for _, a := range adopted {
|
||||
fmt.Fprintf(out, " adopted %s\n", a)
|
||||
}
|
||||
for _, n := range failed {
|
||||
fmt.Fprintf(out, " failed %s\n", n)
|
||||
}
|
||||
if len(failed) > 0 {
|
||||
return fmt.Errorf("cutover: %d peer(s) failed", len(failed))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func adoptAlloc(ctx context.Context, nodeID, allocID string) error {
|
||||
db, err := store.Open(certpaths.DBPath())
|
||||
if err != nil {
|
||||
return fmt.Errorf("open db: %w", err)
|
||||
}
|
||||
defer db.Close()
|
||||
hist := store.NewAllocHistoryRepo(db)
|
||||
if err := hist.EnsureSchema(ctx); err != nil {
|
||||
return fmt.Errorf("alloc history schema: %w", err)
|
||||
}
|
||||
entry := store.AllocHistoryEntry{
|
||||
AllocID: allocID,
|
||||
NodeID: nodeID,
|
||||
FromState: "daemon-managed",
|
||||
ToState: "ssh-push-managed",
|
||||
Timestamp: time.Now().UTC(),
|
||||
Reason: "p14b-cutover",
|
||||
}
|
||||
return hist.Record(ctx, entry)
|
||||
}
|
||||
|
||||
func init() {
|
||||
clusterCutoverCmd.Flags().DurationVar(&cutoverTimeout, "timeout", 5*time.Minute,
|
||||
"max time for the full cutover across all peers")
|
||||
}
|
||||
@@ -0,0 +1,518 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/cluster"
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
"git.cloudinit.dev/coreci/orca/internal/paths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
type mockRotateTransport struct {
|
||||
mu sync.Mutex
|
||||
calls []mockRotateCall
|
||||
written []mockRotateWrite
|
||||
responses []mockRotateResp
|
||||
sticky []mockRotateResp
|
||||
}
|
||||
|
||||
type mockRotateCall struct {
|
||||
peer string
|
||||
cmd string
|
||||
}
|
||||
|
||||
type mockRotateWrite struct {
|
||||
peer string
|
||||
path string
|
||||
content []byte
|
||||
mode os.FileMode
|
||||
}
|
||||
|
||||
type mockRotateResp struct {
|
||||
match string
|
||||
out string
|
||||
exit int
|
||||
}
|
||||
|
||||
func (m *mockRotateTransport) Exec(_ context.Context, peer, cmd string) ([]byte, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.calls = append(m.calls, mockRotateCall{peer: peer, cmd: cmd})
|
||||
for _, r := range m.sticky {
|
||||
if r.match == "" || strings.Contains(cmd, r.match) {
|
||||
if r.exit != 0 {
|
||||
return []byte(r.out), &sshExitErr{code: r.exit}
|
||||
}
|
||||
return []byte(r.out), nil
|
||||
}
|
||||
}
|
||||
for i, r := range m.responses {
|
||||
if r.match == "" || strings.Contains(cmd, r.match) {
|
||||
m.responses = append(m.responses[:i], m.responses[i+1:]...)
|
||||
if r.exit != 0 {
|
||||
return []byte(r.out), &sshExitErr{code: r.exit}
|
||||
}
|
||||
return []byte(r.out), nil
|
||||
}
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockRotateTransport) WriteFileIdempotent(_ context.Context, peer, path string, content []byte, mode os.FileMode) (bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.written = append(m.written, mockRotateWrite{peer: peer, path: path, content: content, mode: mode})
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (m *mockRotateTransport) ReadFile(_ context.Context, peer, path string) ([]byte, error) {
|
||||
return nil, errors.New("not implemented")
|
||||
}
|
||||
|
||||
func (m *mockRotateTransport) countCalls(match string) int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
c := 0
|
||||
for _, call := range m.calls {
|
||||
if strings.Contains(call.cmd, match) {
|
||||
c++
|
||||
}
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func (m *mockRotateTransport) writtenPaths() []string {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
out := make([]string, 0, len(m.written))
|
||||
for _, w := range m.written {
|
||||
out = append(out, w.path)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func cutoverTestNode(t *testing.T, name string) *model.Node {
|
||||
t.Helper()
|
||||
db, err := store.Open(certpaths.DBPath())
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
n := &model.Node{
|
||||
ID: "node-" + name,
|
||||
Name: name,
|
||||
Address: name + ":8443",
|
||||
State: model.NodeStateReady,
|
||||
JoinedAt: time.Now().UTC(),
|
||||
LastSeen: time.Now().UTC(),
|
||||
Kind: string(model.NodeKindLinux),
|
||||
}
|
||||
if err := store.NewNodeRepo(db).Insert(context.Background(), n); err != nil {
|
||||
t.Fatalf("insert node: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func TestCutover_StopsDaemonAndAdoptsAllocs(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "peer-a")
|
||||
cutoverTestNode(t, "peer-b")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("systemctl stop orca-daemon.service", "", 0)
|
||||
mx.queueAlways("list-units", "orca-alloc-web-0.service loaded active running\n", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
clusterCmd.SetOut(&buf)
|
||||
clusterCmd.SetErr(&buf)
|
||||
clusterCutoverCmd.SetOut(&buf)
|
||||
clusterCutoverCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "cutover", "--timeout", "10s"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("cutover: %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "cutover complete") {
|
||||
t.Errorf("expected cutover complete, got: %s", out)
|
||||
}
|
||||
stops := mx.countCalls("systemctl stop orca-daemon.service")
|
||||
if stops != 2 {
|
||||
t.Errorf("expected 2 daemon stops, got %d", stops)
|
||||
}
|
||||
if !strings.Contains(out, "adopted") {
|
||||
t.Errorf("expected adopted in output, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutover_AlreadyStoppedIsIdempotent(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "migrated")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("systemctl stop orca-daemon.service", "", 5)
|
||||
mx.queueAlways("list-units", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
clusterCutoverCmd.SetOut(&buf)
|
||||
clusterCutoverCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "cutover", "--timeout", "5s"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("cutover should be idempotent: %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "already stopped") && !strings.Contains(out, "already-stopped") {
|
||||
t.Errorf("expected already-stopped, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRotateLead_CopiesClusterStateAndRotatesSSHKeys(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
target := cutoverTestNode(t, "newlead")
|
||||
_ = cutoverTestNode(t, "other-peer")
|
||||
|
||||
clusterDir := paths.ClusterDir()
|
||||
if err := os.MkdirAll(clusterDir, 0o755); err != nil {
|
||||
t.Fatalf("mkdir cluster dir: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(certpaths.CACertPath(), []byte("FAKE-CA-CRT"), 0o644); err != nil {
|
||||
t.Fatalf("write ca.crt: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(certpaths.CAKeyPath(), []byte("FAKE-CA-KEY"), 0o600); err != nil {
|
||||
t.Fatalf("write ca.key: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(paths.MasterKeyPath(), []byte("FAKE-MASTER-KEY"), 0o600); err != nil {
|
||||
t.Fatalf("write master.key: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(paths.ConfigPath(), []byte("# orca config"), 0o644); err != nil {
|
||||
t.Fatalf("write config.md: %v", err)
|
||||
}
|
||||
|
||||
mt := &mockRotateTransport{}
|
||||
mt.sticky = append(mt.sticky, mockRotateResp{match: "mkdir -p", out: "", exit: 0})
|
||||
mt.sticky = append(mt.sticky, mockRotateResp{match: "authorized_keys", out: "", exit: 0})
|
||||
driftTransportOverride = mt
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
clusterCmd.SetOut(&buf)
|
||||
clusterCmd.SetErr(&buf)
|
||||
clusterRotateLeadCmd.SetOut(&buf)
|
||||
clusterRotateLeadCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "rotate-lead", "--to", target.Name})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("rotate-lead: %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "lead rotated") {
|
||||
t.Errorf("expected lead rotated, got: %s", out)
|
||||
}
|
||||
written := mt.writtenPaths()
|
||||
if !containsPath(written, "/etc/orca/cluster/ca.key") {
|
||||
t.Errorf("expected ca.key to be copied, written: %v", written)
|
||||
}
|
||||
if !containsPath(written, "/etc/orca/cluster/master.key") {
|
||||
t.Errorf("expected master.key to be copied, written: %v", written)
|
||||
}
|
||||
|
||||
lead, err := readCurrentLead(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("read lead: %v", err)
|
||||
}
|
||||
if lead != target.Name {
|
||||
t.Errorf("lead = %q, want %q", lead, target.Name)
|
||||
}
|
||||
|
||||
newKey, err := os.ReadFile(certpaths.SSHKeyPath())
|
||||
if err != nil {
|
||||
t.Fatalf("read new ssh key: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(newKey), "PRIVATE KEY") {
|
||||
t.Errorf("expected a new private key to be written, got: %s", string(newKey))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRotateLead_ProxmoxTargetRefused(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
db, err := store.Open(certpaths.DBPath())
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
proxmoxNode := &model.Node{
|
||||
ID: "node-prox",
|
||||
Name: "prox-node",
|
||||
Address: "prox-node:8443",
|
||||
State: model.NodeStateReady,
|
||||
JoinedAt: time.Now().UTC(),
|
||||
LastSeen: time.Now().UTC(),
|
||||
Kind: string(model.NodeKindProxmox),
|
||||
}
|
||||
if err := store.NewNodeRepo(db).Insert(context.Background(), proxmoxNode); err != nil {
|
||||
t.Fatalf("insert proxmox node: %v", err)
|
||||
}
|
||||
|
||||
mt := &mockRotateTransport{}
|
||||
driftTransportOverride = mt
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
clusterRotateLeadCmd.SetOut(&buf)
|
||||
clusterRotateLeadCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "rotate-lead", "--to", "prox-node"})
|
||||
err = rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("expected error for Proxmox target, got nil")
|
||||
}
|
||||
if !errors.Is(err, cluster.ErrProxmoxNotLead) && !strings.Contains(err.Error(), "Proxmox") {
|
||||
t.Errorf("expected Proxmox refusal, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRotateLead_AlreadyLeadIsNoop(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
target := cutoverTestNode(t, "currentlead")
|
||||
if err := writeCurrentLead(context.Background(), target.Name); err != nil {
|
||||
t.Fatalf("write lead: %v", err)
|
||||
}
|
||||
|
||||
mt := &mockRotateTransport{}
|
||||
driftTransportOverride = mt
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
clusterRotateLeadCmd.SetOut(&buf)
|
||||
clusterRotateLeadCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "rotate-lead", "--to", target.Name})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("rotate-lead should be no-op when already lead: %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "already") {
|
||||
t.Errorf("expected already-lead message, got: %s", out)
|
||||
}
|
||||
if len(mt.calls) != 0 {
|
||||
t.Errorf("expected no SSH calls for no-op, got: %+v", mt.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoOrcaOnServer_CleanPeerPasses(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "clean-peer")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("command -v orca", "", 0)
|
||||
mx.queueAlways("systemctl list-units", "", 0)
|
||||
mx.queueAlways("pgrep", "", 0)
|
||||
mx.queueAlways("find /etc/orca", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
noOrcaOnServerCmd.SetOut(&buf)
|
||||
noOrcaOnServerCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"doctor", "no-orca-on-server"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("no-orca-on-server (clean): %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "PASS") {
|
||||
t.Errorf("expected PASS for clean peer, got: %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "all peers clean") {
|
||||
t.Errorf("expected all-peers-clean message, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoOrcaOnServer_DirtyPeerBinaryReportsViolation(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "dirty-bin")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("command -v orca", "/usr/local/bin/orca\n", 0)
|
||||
mx.queueAlways("systemctl list-units", "", 0)
|
||||
mx.queueAlways("pgrep", "", 0)
|
||||
mx.queueAlways("find /etc/orca", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
noOrcaOnServerCmd.SetOut(&buf)
|
||||
noOrcaOnServerCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"doctor", "no-orca-on-server"})
|
||||
err := rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("expected error for dirty peer, got nil")
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "FAIL") {
|
||||
t.Errorf("expected FAIL for dirty peer, got: %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "binary in PATH") {
|
||||
t.Errorf("expected binary-in-PATH violation, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoOrcaOnServer_DirtyPeerProcessReportsViolation(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "dirty-proc")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("command -v orca", "", 0)
|
||||
mx.queueAlways("systemctl list-units", "", 0)
|
||||
mx.queueAlways("pgrep", "12345\n", 0)
|
||||
mx.queueAlways("find /etc/orca", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
noOrcaOnServerCmd.SetOut(&buf)
|
||||
noOrcaOnServerCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"doctor", "no-orca-on-server"})
|
||||
err := rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("expected error for dirty peer (process), got nil")
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "process running") {
|
||||
t.Errorf("expected process-running violation, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompatCheck_AllSameVersionPasses(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "peer-a")
|
||||
cutoverTestNode(t, "peer-b")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("orca version", "{\"version\":\""+version+"\"}\n", 0)
|
||||
mx.queueAlways("test -f /etc/orca/cluster/render-contract.json", "", 0)
|
||||
mx.queueAlways("test -d /etc/orca/cluster/txns", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
compatCheckCmd.SetOut(&buf)
|
||||
compatCheckCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "compat-check"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("compat-check (same): %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "all peers compatible") {
|
||||
t.Errorf("expected all-peers-compatible, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompatCheck_MixedCompatiblePasses(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "peer-a")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("orca version", "{\"version\":\"0.1.1\"}\n", 0)
|
||||
mx.queueAlways("test -f /etc/orca/cluster/render-contract.json", "", 0)
|
||||
mx.queueAlways("test -d /etc/orca/cluster/txns", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
compatCheckCmd.SetOut(&buf)
|
||||
compatCheckCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "compat-check"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Logf("output: %s", buf.String())
|
||||
t.Fatalf("compat-check (mixed-compatible): %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompatCheck_IncompatibleReportsIssue(t *testing.T) {
|
||||
_, cleanup := initTestEnv(t)
|
||||
defer cleanup()
|
||||
resetRootFlags(t)
|
||||
|
||||
cutoverTestNode(t, "peer-old")
|
||||
|
||||
mx := &scriptedDrainExec{}
|
||||
mx.queueAlways("orca version", "{\"version\":\"0.8.0\"}\n", 0)
|
||||
mx.queueAlways("test -f /etc/orca/cluster/render-contract.json", "", 0)
|
||||
mx.queueAlways("test -d /etc/orca/cluster/txns", "", 0)
|
||||
drainExecOverride = mx
|
||||
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
compatCheckCmd.SetOut(&buf)
|
||||
compatCheckCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"cluster", "compat-check"})
|
||||
err := rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("expected compat-check to fail for incompatible versions, got nil")
|
||||
}
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "incompatible") {
|
||||
t.Errorf("expected incompatible in output, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func containsPath(paths []string, want string) bool {
|
||||
for _, p := range paths {
|
||||
if p == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func init() {}
|
||||
@@ -98,6 +98,6 @@ var doctorProxmoxCmd = &cobra.Command{
|
||||
}
|
||||
|
||||
func init() {
|
||||
doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd)
|
||||
doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd, noOrcaOnServerCmd)
|
||||
rootCmd.AddCommand(doctorCmd)
|
||||
}
|
||||
|
||||
@@ -72,6 +72,9 @@ func resetCommandFlags() {
|
||||
upgradeTransportOverride = nil
|
||||
peersListerOverride = nil
|
||||
migration.SetCAImporter(nil)
|
||||
cutoverTimeout = 5 * time.Minute
|
||||
rotateLeadTo = ""
|
||||
rotateLeadForce = false
|
||||
resetNSFlags()
|
||||
// Reset per-command output writers so tests that polluted them
|
||||
// (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into
|
||||
@@ -80,6 +83,7 @@ func resetCommandFlags() {
|
||||
nodeDrainCmd, daemonCmd, daemonDrainAndStopCmd,
|
||||
jobCmd, jobMigrateCmd, jobRunCmd, jobListCmd, jobStopCmd, jobLogsCmd, jobLintCmd, jobVerifyCmd,
|
||||
logsCmd,
|
||||
clusterCmd, clusterCutoverCmd, clusterRotateLeadCmd, compatCheckCmd, noOrcaOnServerCmd,
|
||||
} {
|
||||
if c != nil {
|
||||
c.SetOut(nil)
|
||||
|
||||
@@ -0,0 +1,338 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"crypto/rand"
|
||||
"encoding/pem"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
"golang.org/x/crypto/ssh"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/cluster"
|
||||
"git.cloudinit.dev/coreci/orca/internal/engine"
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
"git.cloudinit.dev/coreci/orca/internal/paths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
var (
|
||||
rotateLeadTo string
|
||||
rotateLeadForce bool
|
||||
rotateLeadDebug bool
|
||||
)
|
||||
|
||||
var clusterRotateLeadCmd = &cobra.Command{
|
||||
Use: "rotate-lead --to <new-lead-host>",
|
||||
Short: "Rotate the cluster lead to a new bare Linux node (REQ-114, R-003)",
|
||||
Long: `Rotate the cluster lead to a new bare Linux node (REQ-114).
|
||||
|
||||
Steps:
|
||||
1. Verify the new lead is a registered bare Linux node (R-003:
|
||||
Proxmox nodes are permanently ineligible — hypervisor kernel is
|
||||
shared with guests).
|
||||
2. Copy the cluster CA (ca.crt + ca.key), master.key, config.md,
|
||||
and the transaction log to the new lead via SSH.
|
||||
3. Update the local cluster state to point at the new lead.
|
||||
4. Workloads keep running — peer certs are already distributed.
|
||||
5. Rotate the SSH keypair: generate a new Ed25519 key, deploy the
|
||||
public key to every peer's authorized_keys, and deprecate the
|
||||
old key.
|
||||
6. Idempotent: if the new lead is already the current lead, no-op.
|
||||
|
||||
--force skips the R-003 verification (use with caution).`,
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return runRotateLead(cmd)
|
||||
},
|
||||
}
|
||||
|
||||
func runRotateLead(cmd *cobra.Command) error {
|
||||
if rotateLeadTo == "" {
|
||||
return fmt.Errorf("--to is required")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
log := newLogger()
|
||||
|
||||
currentLead, err := readCurrentLead(ctx)
|
||||
if err != nil {
|
||||
log.Warn("rotate-lead: cannot read current lead", "error", err)
|
||||
}
|
||||
if currentLead == rotateLeadTo {
|
||||
if jsonOutput {
|
||||
return printJSON(map[string]any{
|
||||
"already_lead": true,
|
||||
"lead": rotateLeadTo,
|
||||
})
|
||||
}
|
||||
fmt.Fprintf(cmd.OutOrStdout(), "✓ %s is already the cluster lead; no-op\n", rotateLeadTo)
|
||||
return nil
|
||||
}
|
||||
|
||||
reg, closer, err := nodeRegistry()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer closer()
|
||||
|
||||
nodes, err := reg.List(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list nodes: %w", err)
|
||||
}
|
||||
|
||||
if !rotateLeadForce {
|
||||
nodeInfos := make([]cluster.NodeInfo, 0, len(nodes))
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
kind := cluster.NodeKindLinux
|
||||
if n.Kind == string(model.NodeKindProxmox) {
|
||||
kind = cluster.NodeKindProxmox
|
||||
}
|
||||
nodeInfos = append(nodeInfos, cluster.NodeInfo{Hostname: n.Name, Kind: kind})
|
||||
}
|
||||
if err := cluster.ValidateLeadRotation(rotateLeadTo, nodeInfos); err != nil {
|
||||
return fmt.Errorf("rotate-lead: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
target, err := findNode(ctx, reg, rotateLeadTo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
peer := peerAddrForNode(target)
|
||||
if peer == "" {
|
||||
return fmt.Errorf("cannot resolve SSH address for target node %q", target.Name)
|
||||
}
|
||||
|
||||
transport, err := driftTransportFromCtx()
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh transport: %w", err)
|
||||
}
|
||||
|
||||
copyResult, err := copyClusterStateToNewLead(ctx, transport, peer)
|
||||
if err != nil {
|
||||
return fmt.Errorf("rotate-lead: copy cluster state: %w", err)
|
||||
}
|
||||
|
||||
if err := writeCurrentLead(ctx, target.Name); err != nil {
|
||||
return fmt.Errorf("rotate-lead: update cluster state: %w", err)
|
||||
}
|
||||
|
||||
rotateResult, err := rotateSSHKeys(ctx, transport, nodes)
|
||||
if err != nil {
|
||||
log.Warn("rotate-lead: SSH key rotation partial", "error", err)
|
||||
}
|
||||
|
||||
result := map[string]any{
|
||||
"old_lead": currentLead,
|
||||
"new_lead": target.Name,
|
||||
"peer": peer,
|
||||
"copied": copyResult,
|
||||
"ssh_key_rotation": rotateResult,
|
||||
}
|
||||
|
||||
if db, dbErr := store.Open(certpaths.DBPath()); dbErr == nil {
|
||||
engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, "cli", "cluster.rotate_lead", target.Name, "success", nil, result)
|
||||
db.Close()
|
||||
}
|
||||
|
||||
if jsonOutput {
|
||||
return printJSON(result)
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
fmt.Fprintf(out, "✓ lead rotated to %s\n", target.Name)
|
||||
for _, f := range copyResult {
|
||||
fmt.Fprintf(out, " copied %s\n", f)
|
||||
}
|
||||
if rotateResult != nil {
|
||||
fmt.Fprintf(out, " rotated ssh key (deployed to %d peer(s), deprecated old key)\n", rotateResult.Deployed)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func copyClusterStateToNewLead(ctx context.Context, transport driftTransport, peer string) ([]string, error) {
|
||||
files := []struct {
|
||||
src string
|
||||
dst string
|
||||
}{
|
||||
{certpaths.CACertPath(), "/etc/orca/cluster/ca.crt"},
|
||||
{certpaths.CAKeyPath(), "/etc/orca/cluster/ca.key"},
|
||||
{paths.MasterKeyPath(), "/etc/orca/cluster/master.key"},
|
||||
{paths.ConfigPath(), "/etc/orca/cluster/config.md"},
|
||||
}
|
||||
var copied []string
|
||||
for _, f := range files {
|
||||
content, err := os.ReadFile(f.src)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
return copied, fmt.Errorf("read %s: %w", f.src, err)
|
||||
}
|
||||
mkdirCmd := fmt.Sprintf("mkdir -p %s", filepath.Dir(f.dst))
|
||||
if _, err := transport.Exec(ctx, peer, mkdirCmd); err != nil {
|
||||
return copied, fmt.Errorf("mkdir on new lead for %s: %w", f.dst, err)
|
||||
}
|
||||
if _, err := transport.WriteFileIdempotent(ctx, peer, f.dst, content, 0o600); err != nil {
|
||||
return copied, fmt.Errorf("write %s: %w", f.dst, err)
|
||||
}
|
||||
copied = append(copied, f.dst)
|
||||
}
|
||||
|
||||
txnDir := paths.TxnDir()
|
||||
if entries, err := os.ReadDir(txnDir); err == nil {
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() {
|
||||
continue
|
||||
}
|
||||
localDir := filepath.Join(txnDir, e.Name())
|
||||
if err := copyTxnDir(ctx, transport, peer, localDir, e.Name()); err != nil {
|
||||
slog.Warn("rotate-lead: copy txn dir failed",
|
||||
slog.String("txn", e.Name()), "error", err)
|
||||
continue
|
||||
}
|
||||
copied = append(copied, "txns/"+e.Name())
|
||||
}
|
||||
}
|
||||
return copied, nil
|
||||
}
|
||||
|
||||
func copyTxnDir(ctx context.Context, transport driftTransport, peer, localDir, txnID string) error {
|
||||
files := []string{"manifest.json", "manifest.sig", "desired-state.json", "apply.sh", "verify.sh", "rollback.sh"}
|
||||
remoteDir := "/etc/orca/cluster/txns/" + txnID
|
||||
mkdirCmd := fmt.Sprintf("mkdir -p %s", remoteDir)
|
||||
if _, err := transport.Exec(ctx, peer, mkdirCmd); err != nil {
|
||||
return fmt.Errorf("mkdir %s: %w", remoteDir, err)
|
||||
}
|
||||
for _, f := range files {
|
||||
p := filepath.Join(localDir, f)
|
||||
content, err := os.ReadFile(p)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
dst := remoteDir + "/" + f
|
||||
if _, err := transport.WriteFileIdempotent(ctx, peer, dst, content, 0o644); err != nil {
|
||||
return fmt.Errorf("write %s: %w", dst, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type rotateSSHKeysResult struct {
|
||||
Deployed int `json:"deployed"`
|
||||
Failed []string `json:"failed,omitempty"`
|
||||
OldKeyHash string `json:"old_key_hash,omitempty"`
|
||||
}
|
||||
|
||||
func rotateSSHKeys(ctx context.Context, transport driftTransport, nodes []*model.Node) (*rotateSSHKeysResult, error) {
|
||||
pubPath := certpaths.SSHPubPath()
|
||||
keyPath := certpaths.SSHKeyPath()
|
||||
|
||||
oldPub, _ := os.ReadFile(pubPath)
|
||||
|
||||
newPriv, newPub, err := generateEd25519Keypair()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("generate new ssh key: %w", err)
|
||||
}
|
||||
if err := os.WriteFile(keyPath, newPriv, 0o600); err != nil {
|
||||
return nil, fmt.Errorf("write new ssh key: %w", err)
|
||||
}
|
||||
if err := os.WriteFile(pubPath, newPub, 0o644); err != nil {
|
||||
return nil, fmt.Errorf("write new ssh pub: %w", err)
|
||||
}
|
||||
|
||||
res := &rotateSSHKeysResult{Failed: []string{}}
|
||||
for i := range nodes {
|
||||
n := nodes[i]
|
||||
peer := peerAddrForNode(n)
|
||||
if peer == "" {
|
||||
continue
|
||||
}
|
||||
deployCmd := fmt.Sprintf("mkdir -p ~/.ssh && echo %s >> ~/.ssh/authorized_keys && chmod 600 ~/.ssh/authorized_keys", sshQuote(strings.TrimSpace(string(newPub))))
|
||||
if _, err := transport.Exec(ctx, peer, deployCmd); err != nil {
|
||||
res.Failed = append(res.Failed, n.Name)
|
||||
continue
|
||||
}
|
||||
res.Deployed++
|
||||
}
|
||||
|
||||
if len(oldPub) > 0 {
|
||||
res.OldKeyHash = sshFingerprint(oldPub)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func generateEd25519Keypair() (privBytes []byte, pubBytes []byte, err error) {
|
||||
pubKey, privKey, err := ed25519.GenerateKey(rand.Reader)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
sshPub, err := ssh.NewPublicKey(pubKey)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
pubBytes = ssh.MarshalAuthorizedKey(sshPub)
|
||||
pemBlock, err := ssh.MarshalPrivateKey(privKey, "")
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
privBytes = pem.EncodeToMemory(pemBlock)
|
||||
return privBytes, pubBytes, nil
|
||||
}
|
||||
|
||||
func sshFingerprint(pub []byte) string {
|
||||
pk, _, _, _, err := ssh.ParseAuthorizedKey(pub)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return ssh.FingerprintSHA256(pk)
|
||||
}
|
||||
|
||||
func readCurrentLead(ctx context.Context) (string, error) {
|
||||
leadPath := filepath.Join(paths.ClusterDir(), "lead")
|
||||
b, err := os.ReadFile(leadPath)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return "", nil
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
return trimSpace(string(b)), nil
|
||||
}
|
||||
|
||||
func writeCurrentLead(ctx context.Context, name string) error {
|
||||
dir := paths.ClusterDir()
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
leadPath := filepath.Join(dir, "lead")
|
||||
return os.WriteFile(leadPath, []byte(name), 0o644)
|
||||
}
|
||||
|
||||
func trimSpace(s string) string {
|
||||
for len(s) > 0 && (s[0] == ' ' || s[0] == '\t' || s[0] == '\n' || s[0] == '\r') {
|
||||
s = s[1:]
|
||||
}
|
||||
for len(s) > 0 && (s[len(s)-1] == ' ' || s[len(s)-1] == '\t' || s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
|
||||
s = s[:len(s)-1]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func init() {
|
||||
clusterRotateLeadCmd.Flags().StringVar(&rotateLeadTo, "to", "", "new lead host (required, must be a bare Linux node)")
|
||||
clusterRotateLeadCmd.Flags().BoolVar(&rotateLeadForce, "force", false, "skip R-003 verification (use with caution)")
|
||||
_ = rotateLeadDebug
|
||||
}
|
||||
Reference in New Issue
Block a user