pve-confsync/internal/engine/apply.go

package engine

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"time"

	"pve-confsync/internal/adapter"
	"pve-confsync/internal/model"
	"pve-confsync/internal/pve"
)

// Step is the outcome of one API action, for the execution journal.
type Step struct {
	At     time.Time `json:"at"`
	Action string    `json:"action"`
	Error  string    `json:"error,omitempty"`
}

// Result of an apply on one client.
type Result struct {
	Client string `json:"client"`
	Steps  []Step `json:"steps"`
	Err    error  `json:"-"`

	notify func(Step)
}

func (r *Result) log(action string, err error) {
	s := Step{At: time.Now().UTC(), Action: action}
	if err != nil {
		s.Error = err.Error()
	}
	r.Steps = append(r.Steps, s)
	if r.notify != nil {
		r.notify(s)
	}
}

// TaskPoll is the poll interval of the SDN reload task.
var TaskPoll = 2 * time.Second

// Apply executes a plan. Non SDN changes are applied directly; SDN changes are
// written under the global SDN lock (pve-network >= 1.1.3, PVE 9.0), then
// committed with PUT /cluster/sdn. Any failure before the commit rolls the
// pending SDN changes back (POST /cluster/sdn/rollback).
func Apply(ctx context.Context, c *pve.Client, plan Plan) Result {
	return ApplyWithLog(ctx, c, plan, nil)
}

// ApplyWithLog is Apply calling notify after each step (live journal of the web UI).
func ApplyWithLog(ctx context.Context, c *pve.Client, plan Plan, notify func(Step)) Result {
	res := Result{Client: plan.Client, notify: notify}
	if len(plan.Errors) > 0 {
		res.Err = fmt.Errorf("plan has %d error(s), nothing applied", len(plan.Errors))
		res.log("plan check", res.Err)
		return res
	}
	if _, err := c.CheckVersion(ctx); err != nil {
		res.Err = err
		res.log("version check", err)
		return res
	}
	var sdn []Change
	for _, ch := range plan.Changes {
		a, _ := adapter.ByKind(ch.Kind)
		if a.SDN() {
			sdn = append(sdn, ch)
			continue
		}
		err := run(ctx, c, a, ch, "")
		res.log(describe(ch), err)
		if err != nil {
			res.Err = err
			return res
		}
	}
	if len(sdn) == 0 {
		return res
	}

	var token string
	err := c.Post(ctx, "cluster/sdn/lock", nil, &token)
	res.log("sdn lock", err)
	if err != nil {
		res.Err = fmt.Errorf("SDN lock (pending changes on the client?): %w", err)
		return res
	}
	rollback := func(cause error) {
		f := url.Values{"lock-token": {token}, "release-lock": {"1"}}
		rerr := c.Post(ctx, "cluster/sdn/rollback", f, nil)
		res.log("sdn rollback", rerr)
		res.Err = cause
		if rerr != nil {
			res.Err = errors.Join(cause, fmt.Errorf("rollback failed, lock still held: %w", rerr))
		}
	}
	for _, ch := range sdn {
		a, _ := adapter.ByKind(ch.Kind)
		err := run(ctx, c, a, ch, token)
		res.log(describe(ch), err)
		if err != nil {
			rollback(err)
			return res
		}
	}
	var upid string
	err = c.Put(ctx, "cluster/sdn", url.Values{"lock-token": {token}, "release-lock": {"1"}}, &upid)
	res.log("sdn apply", err)
	if err != nil {
		rollback(err)
		return res
	}
	// The commit is done: from here a failure is in the network reload of the
	// nodes, the running configuration is already the new one.
	err = c.WaitTask(ctx, upid, TaskPoll)
	res.log("sdn reload task "+upid, err)
	if err != nil {
		res.Err = fmt.Errorf("SDN configuration committed but reload failed: %w", err)
	}
	return res
}

func run(ctx context.Context, c *pve.Client, a adapter.Adapter, ch Change, lock string) error {
	switch ch.Op {
	case OpCreate:
		return a.Create(ctx, c, model.Object{Kind: ch.Kind, ID: ch.ID, Fields: ch.Set}, lock)
	case OpUpdate:
		return a.Update(ctx, c, ch.Target, ch.Set, ch.Unset, lock)
	case OpDelete:
		return a.Delete(ctx, c, ch.Target, lock)
	}
	return fmt.Errorf("unknown op %q", ch.Op)
}

func describe(ch Change) string {
	return fmt.Sprintf("%s %s/%s", ch.Op, ch.Kind, ch.ID)
}

// Verify re-reads the client and checks that nothing is left to do and that
// no node reports an SDN zone in error (GET /nodes/{node}/sdn/zones).
func Verify(ctx context.Context, c *pve.Client, plan func(actual []model.Object) Plan, kinds []model.Kind) error {
	snap, err := Read(ctx, c, kinds, adapter.ReadCurrent)
	if err != nil {
		return err
	}
	var errs []error
	if p := plan(snap.Objects); !p.Empty() {
		for _, ch := range p.Changes {
			errs = append(errs, fmt.Errorf("still different after apply: %s", describe(ch)))
		}
	}
	var status []map[string]any
	if err := c.Get(ctx, "cluster/status", nil, &status); err != nil {
		return errors.Join(append(errs, err)...)
	}
	for _, s := range status {
		if pve.Str(s["type"]) != "node" || pve.Str(s["online"]) != "1" {
			continue
		}
		node := pve.Str(s["name"])
		var zones []map[string]any
		if err := c.Get(ctx, "nodes/"+url.PathEscape(node)+"/sdn/zones", nil, &zones); err != nil {
			errs = append(errs, err)
			continue
		}
		for _, z := range zones {
			if pve.Str(z["status"]) == "error" {
				errs = append(errs, fmt.Errorf("node %s: SDN zone %s in error", node, pve.Str(z["zone"])))
			}
		}
	}
	return errors.Join(errs...)
}