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...) }