pve-confsync/internal/nodesync/apply.go

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 <mode> <owner> <hash> <path>" then the content in base64 on one line,
// "D - - <hash> <path>" (delete) or "B - - - <path>" (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"
}