package nodesync import ( "bytes" "context" "encoding/base64" "errors" "fmt" "net/url" "sort" "strings" "time" "pve-confsync/internal/pve" ) // Step is the outcome of one action, for the execution journal. type Step struct { At time.Time `json:"at"` Action string `json:"action"` Error string `json:"error,omitempty"` } // Timing of an apply; variables for the tests. var ( // RestoreDelay: when the network was reloaded, the node restores its // previous configuration by itself after this delay unless pve-confsync // reaches it again over SSH and cancels the restore. RestoreDelay = 120 * time.Second // ProbeTries and ProbeWait: attempts to reach the node after the reload. ProbeTries = 6 ProbeWait = 5 * time.Second TaskPoll = 2 * time.Second ) // BackupRoot is where the previous files are kept on the node, one folder // per apply, with the restore.sh script that puts them back. const BackupRoot = "/var/lib/pve-confsync/backup" // Apply writes a plan on a target node: // 1. one SSH script checks that every file is still the one of the dry-run, // backs up the files (and /etc/network/interfaces, /etc/resolv.conf when // the API writes them), writes restore.sh, then writes the new files // atomically (temporary file then mv, mode and owner kept); // 2. the API writes the interfaces (in interfaces.new) and the DNS servers; // 3. when the network is reloaded, a systemd timer is armed on the node: it // runs restore.sh and ifreload -a after RestoreDelay; // 4. the reload commands run (sysctl, chrony, multipath, network last); // 5. the node must answer again over SSH and the API: the timer is stopped. // // A failure before the network reload restores the backups (and drops // interfaces.new); a reload failure restores them and runs the reloads again. func Apply(ctx context.Context, ep Endpoint, plan Plan, notify func(Step)) error { log := func(action string, err error) error { s := Step{At: time.Now().UTC(), Action: action} if err != nil { s.Error = err.Error() } if notify != nil { notify(s) } return err } if len(plan.Errors) > 0 { return log("plan check", fmt.Errorf("plan has %d error(s), nothing applied", len(plan.Errors))) } if plan.Empty() { return nil } if ep.Shell == nil { return log("ssh", fmt.Errorf("%s: SSH host key not recorded yet, test the node first", ep.Key)) } if _, err := NodeVersion(ctx, ep); err != nil { return log("version check", err) } stamp := time.Now().UTC().Format("20060102-150405") dir := BackupRoot + "/" + stamp unit := "pve-confsync-restore-" + stamp var netAPI, dns bool for _, c := range plan.Changes { netAPI = netAPI || c.Item == "network" dns = dns || c.Item == "dns" } if _, err := ep.Shell.Run(ctx, prepareScript(dir), prepareInput(plan, netAPI, dns)); err != nil { return log("check, back up and write the files ("+dir+")", err) } log("check, back up and write the files ("+dir+")", nil) restore := func(cause error) error { _, rerr := ep.Shell.Run(ctx, restoreScript(dir), nil) log("restore the previous files", rerr) if rerr != nil { return errors.Join(cause, fmt.Errorf("restore failed, run sh %s/restore.sh on the node: %w", dir, rerr)) } return cause } for _, c := range plan.Changes { if c.Item != "network" && c.Item != "dns" { continue } if err := log(describe(ep, c), writeAPI(ctx, ep, c)); err != nil { return restore(err) } } network := plan.Network() if network { _, err := ep.Shell.Run(ctx, armScript(dir, unit), nil) if err := log(fmt.Sprintf("arm the automatic restore (%s, %s)", unit, RestoreDelay), err); err != nil { return restore(err) } } var done []string // reload commands already run for _, cmd := range plan.Reloads() { var err error action := cmd if cmd == "ifreload -a" && netAPI { action = "PUT /nodes/" + ep.Node + "/network (ifreload -a)" err = reloadNetworkAPI(ctx, ep) } else { _, err = ep.Shell.Run(ctx, reloadScript(cmd), nil) } done = append(done, cmd) if log("reload: "+action, err) == nil { continue } cause := fmt.Errorf("reload %q: %w", cmd, err) if network && !reachable(ctx, ep) { return log("node unreachable", fmt.Errorf("%w; the node is unreachable: it restores its previous configuration by itself within %s", cause, RestoreDelay)) } // back to the previous files, which the reloads put in effect again err = restore(cause) for _, c := range done { _, rerr := ep.Shell.Run(ctx, reloadScript(c), nil) log("reload after restore: "+c, rerr) } if network { _, derr := ep.Shell.Run(ctx, disarmScript(unit), nil) log("stop the automatic restore", derr) } return err } if network { if !reachable(ctx, ep) { return log("reach the node again", fmt.Errorf("%s unreachable after the network reload: it restores its previous configuration by itself within %s", ep.Key, RestoreDelay)) } log("reach the node again (SSH and API)", nil) _, err := ep.Shell.Run(ctx, disarmScript(unit), nil) if err := log("stop the automatic restore", err); err != nil { return fmt.Errorf("stop the automatic restore (systemctl stop %s.timer on the node): %w", unit, err) } } return nil } func describe(ep Endpoint, c Change) string { if c.Item == "dns" { return "PUT /nodes/" + ep.Node + "/dns" } return fmt.Sprintf("%s %s/%s", c.Op, c.Item, c.ID) } // reachable opens a new SSH connection and calls the API, a few times. func reachable(ctx context.Context, ep Endpoint) bool { for i := 0; i < ProbeTries; i++ { if i > 0 { select { case <-ctx.Done(): return false case <-time.After(ProbeWait): } } sh := ep.Shell if ep.Dial != nil { var err error if sh, err = ep.Dial(); err != nil { continue } } if _, err := sh.Run(ctx, "# confsync:probe\ntrue", nil); err != nil { continue } if _, err := NodeVersion(ctx, ep); err == nil { return true } } return false } func nodePath(ep Endpoint, rest string) string { return "nodes/" + url.PathEscape(ep.Node) + "/" + rest } // writeAPI writes one interface (POST, PUT with every field, DELETE) or the // DNS servers (PUT with every server). func writeAPI(ctx context.Context, ep Endpoint, c Change) error { if c.Item == "dns" { f := url.Values{} for k, v := range c.Full { f.Set(k, pve.Str(v)) } return ep.API.Put(ctx, nodePath(ep, "dns"), f, nil) } switch c.Op { case OpCreate: f := form(c.Full) f.Set("iface", c.ID) return ep.API.Post(ctx, nodePath(ep, "network"), f, nil) case OpUpdate: f := form(c.Full) if len(c.Unset) > 0 { f.Set("delete", strings.Join(c.Unset, ",")) } return ep.API.Put(ctx, nodePath(ep, "network/"+url.PathEscape(c.ID)), f, nil) case OpDelete: return ep.API.Delete(ctx, nodePath(ep, "network/"+url.PathEscape(c.ID)), nil, nil) } return fmt.Errorf("unknown op %q", c.Op) } func form(fields map[string]any) url.Values { f := url.Values{} keys := make([]string, 0, len(fields)) for k := range fields { keys = append(keys, k) } sort.Strings(keys) for _, k := range keys { f.Set(k, pve.Str(fields[k])) } return f } // reloadNetworkAPI applies interfaces.new (PUT /nodes/{node}/network: rename // then ifreload -a in a task) and waits for the task. func reloadNetworkAPI(ctx context.Context, ep Endpoint) error { var upid string if err := ep.API.Put(ctx, nodePath(ep, "network"), url.Values{}, &upid); err != nil { return err } return ep.API.WaitTask(ctx, upid, TaskPoll) } // ---- scripts run on the node (the first line is a marker for the fake node // of the tests) ---- // prepareInput is the plan given to prepareScript on its standard input: // "W " then the content in base64 on one line, // "D - - " (delete) or "B - - - " (backup only). func prepareInput(plan Plan, netAPI, dns bool) []byte { var b bytes.Buffer if netAPI { b.WriteString("B - - - /etc/network/interfaces\nB - - - /etc/network/interfaces.new\n") } if dns { b.WriteString("B - - - /etc/resolv.conf\n") } for _, c := range plan.Changes { switch { case c.Item == "network" || c.Item == "dns": case c.Op == OpDelete: fmt.Fprintf(&b, "D - - %s %s\n", c.Hash, c.ID) default: fmt.Fprintf(&b, "W %s %s %s %s\n%s\n", c.Mode, c.Owner, c.Hash, c.ID, base64.StdEncoding.EncodeToString(c.Content)) } } return b.Bytes() } func prepareScript(dir string) string { return prepareScriptIn(dir, "") } // prepareScriptIn prefixes every path with root (empty on a node): the tests // run the script for real in a temporary folder. func prepareScriptIn(dir, root string) string { return `# confsync:prepare ` + dir + ` set -u R='` + root + `' B="$R` + dir + `" umask 077 mkdir -p "$B/files" || exit 1 cat > "$B/plan" || exit 1 sum() { if [ -e "$1" ]; then sha256sum < "$1" | cut -d' ' -f1; else echo absent; fi; } # 1. every file must be the one read by the dry-run while read -r op mode owner h p; do case "$op" in W|D) ;; *) continue ;; esac [ "$(sum "$R$p")" = "$h" ] || { echo "changed since the dry-run: $p" >&2; exit 3; } done < "$B/plan" # 2. backups and restore script : > "$B/restore.sh" while read -r op mode owner h p; do case "$op" in W|D|B) ;; *) continue ;; esac if [ -e "$R$p" ]; then mkdir -p "$B/files$(dirname "$p")" && cp -p "$R$p" "$B/files$p" || exit 1 printf "cp -p '%s' '%s'\n" "$B/files$p" "$R$p" >> "$B/restore.sh" else printf "rm -f '%s'\n" "$R$p" >> "$B/restore.sh" fi done < "$B/plan" # 3. writes: temporary file in the same folder, then mv while read -r op mode owner h p; do case "$op" in W) read -r data mkdir -p "$(dirname "$R$p")" || exit 4 t=$(mktemp "$R$p.confsync.XXXXXX") || exit 4 if printf '%s' "$data" | base64 -d > "$t" && chmod "$mode" "$t" && chown "$owner" "$t" && mv -f "$t" "$R$p"; then :; else rm -f "$t"; sh "$B/restore.sh"; echo "write failed: $p" >&2; exit 4 fi ;; D) rm -f "$R$p" || { sh "$B/restore.sh"; echo "delete failed: $p" >&2; exit 4; } ;; esac done < "$B/plan" echo ok ` } func restoreScript(dir string) string { return "# confsync:restore " + dir + "\nsh '" + dir + "/restore.sh'\n" } func armScript(dir, unit string) string { return fmt.Sprintf("# confsync:arm %s\nsystemd-run --quiet --unit=%s --on-active=%ds --timer-property=AccuracySec=1s /bin/sh -c 'sh %s/restore.sh; ifreload -a'\n", unit, unit, int(RestoreDelay.Seconds()), dir) } func disarmScript(unit string) string { return "# confsync:disarm " + unit + "\nsystemctl stop " + unit + ".timer\n" } func reloadScript(cmd string) string { return "# confsync:reload " + cmd + "\n" + cmd + "\n" }