// Package service puts the engine together for the web UI: it reads the // reference and the clients, computes the plans and describes every object // field by field (reference, client mapping, current value, target value). package service import ( "context" "fmt" "sort" "sync" "time" "pve-confsync/internal/adapter" "pve-confsync/internal/config" "pve-confsync/internal/engine" "pve-confsync/internal/model" "pve-confsync/internal/pve" ) // Node is the PVE version of one node. type Node struct { Name string `json:"name"` Version string `json:"version"` } // ClusterInfo is what was read from one cluster. type ClusterInfo struct { Name string `json:"name"` URL string `json:"url"` Version string `json:"version,omitempty"` Nodes []Node `json:"nodes,omitempty"` Objects int `json:"objects"` Error string `json:"error,omitempty"` Warnings []string `json:"warnings,omitempty"` ReadAt time.Time `json:"readAt"` Millis int64 `json:"ms"` } // Field compares one field of an object. type Field struct { Name string `json:"name"` Policy adapter.Policy `json:"policy"` Ref any `json:"ref,omitempty"` // value on the reference Mapping any `json:"mapping,omitempty"` // value of the client mapping Current any `json:"current,omitempty"` // value on the client Target any `json:"target,omitempty"` // value after a synchronization Change string `json:"change,omitempty"` // add, set, unset or empty } // Row statuses. const ( StatusSame = "same" // in line with the reference StatusCreate = "create" // missing on the client StatusUpdate = "update" // different StatusDelete = "delete" // absent from the reference, deleted (prune) StatusExtra = "extra" // absent from the reference, kept (no prune) StatusError = "error" // blocking error (missing mapping, immutable field) ) // Row is one object of a client. type Row struct { Kind model.Kind `json:"kind"` ID string `json:"id"` Key string `json:"key"` Status string `json:"status"` SDN bool `json:"sdn"` Fields []Field `json:"fields"` Errors []string `json:"errors,omitempty"` } // Client is the state of one client: what it has, its plan and its rows. type Client struct { ClusterInfo Plan engine.Plan `json:"plan"` Digest string `json:"digest"` Rows []Row `json:"rows"` Counts map[string]int `json:"counts"` } // State is a full comparison of the reference with its clients. type State struct { At time.Time `json:"at"` Reference ClusterInfo `json:"reference"` Clients []Client `json:"clients"` Kinds []model.Kind `json:"kinds"` Prune bool `json:"prune"` ref []model.Object actuals map[string][]model.Object cfg *config.Config } // Analyze reads the reference (applied SDN configuration) and the clients // (current configuration, in parallel), and compares them. only limits the // clients (empty: all). A cluster that cannot be read is reported in its // Error field, it does not stop the others. func Analyze(ctx context.Context, cfg *config.Config, only []string) *State { st := &State{At: time.Now().UTC(), Kinds: cfg.Kinds, Prune: cfg.Prune, cfg: cfg, actuals: map[string][]model.Object{}} var refSnap model.Snapshot st.Reference, refSnap = read(ctx, cfg.Reference, cfg.Kinds, adapter.ReadRunning) st.ref = refSnap.Objects want := map[string]bool{} for _, n := range only { want[n] = true } var wg sync.WaitGroup var mu sync.Mutex for i := range cfg.Clients { cl := &cfg.Clients[i] if len(only) > 0 && !want[cl.Name] { continue } st.Clients = append(st.Clients, Client{ClusterInfo: ClusterInfo{Name: cl.Name, URL: cl.URL}}) idx := len(st.Clients) - 1 wg.Add(1) go func() { defer wg.Done() info, snap := read(ctx, cl.Cluster, cfg.Kinds, adapter.ReadCurrent) mu.Lock() defer mu.Unlock() st.Clients[idx].ClusterInfo = info if info.Error == "" { st.actuals[cl.Name] = snap.Objects } }() } wg.Wait() for i := range st.Clients { c := &st.Clients[i] c.Counts = map[string]int{} if c.Error != "" || st.Reference.Error != "" { continue } cl, _ := cfg.Client(c.Name) c.Plan = engine.MakePlan(cfg, cl, st.ref, st.actuals[c.Name]) c.Plan.Warnings = append(c.Plan.Warnings, c.Warnings...) c.Digest = c.Plan.Digest() c.Rows = rows(cfg, cl, st.ref, st.actuals[c.Name], c.Plan) for _, r := range c.Rows { c.Counts[r.Status]++ } } return st } func read(ctx context.Context, cl config.Cluster, kinds []model.Kind, mode adapter.ReadMode) (ClusterInfo, model.Snapshot) { info := ClusterInfo{Name: cl.Name, URL: cl.URL, ReadAt: time.Now().UTC()} start := time.Now() defer func() { info.Millis = time.Since(start).Milliseconds() }() c, err := cl.PVE() if err != nil { info.Error = err.Error() return info, model.Snapshot{} } snap, err := engine.Read(ctx, c, kinds, mode) if err != nil { info.Error = err.Error() } info.Version = snap.PVEVersion if vs, verr := c.Versions(ctx); verr == nil { info.Nodes = nodes(vs) } info.Objects = len(snap.Objects) info.Warnings = snap.Warnings info.Millis = time.Since(start).Milliseconds() return info, snap } // rows describes every object of the reference and of the client, field by field. func rows(cfg *config.Config, cl *config.Client, ref, actual []model.Object, plan engine.Plan) []Row { changes := map[string]engine.Change{} for _, c := range plan.Changes { changes[c.Key()] = c } errs := map[string][]string{} for _, e := range plan.Errors { if k, ok := engine.ErrorKey(e); ok { errs[k] = append(errs[k], e) } } cur := map[string]model.Object{} for _, o := range actual { cur[o.Key()] = o } var out []Row seen := map[string]bool{} add := func(k model.Kind, id string, r *model.Object, c *model.Object) { a, err := adapter.ByKind(k) if err != nil { return } key := string(k) + "/" + id seen[key] = true row := Row{Kind: k, ID: id, Key: key, SDN: a.SDN(), Errors: errs[key]} ch, hasChange := changes[key] switch { case len(row.Errors) > 0: row.Status = StatusError case hasChange: row.Status = string(ch.Op) case r == nil: row.Status = StatusExtra default: row.Status = StatusSame } var mapping map[string]any if cl != nil { mapping = model.Normalize(cl.Mapping[k][id]) a.Canonicalize(mapping) } names := map[string]bool{} for _, o := range []*model.Object{r, c} { if o != nil { for f := range o.Fields { names[f] = true } } } for f := range mapping { names[f] = true } for f := range ch.Set { names[f] = true } list := make([]string, 0, len(names)) for f := range names { list = append(list, f) } sort.Strings(list) unset := map[string]bool{} for _, f := range ch.Unset { unset[f] = true } for _, f := range list { fd := Field{Name: f, Policy: cfg.Policy(cl, a, f), Mapping: mapping[f]} if r != nil { fd.Ref = r.Fields[f] } if c != nil { fd.Current = c.Fields[f] } switch { case ch.Op == OpDelete: fd.Change = "unset" case unset[f]: fd.Change = "unset" case ch.Set[f] != nil: fd.Target = ch.Set[f] if fd.Current != nil { fd.Change = "set" } else { fd.Change = "add" } case fd.Policy != adapter.Ignore && fd.Current != nil: fd.Target = fd.Current } row.Fields = append(row.Fields, fd) } out = append(out, row) } for i := range ref { r := &ref[i] if !cfg.Selected(r.Kind, r.ID) { continue } var c *model.Object if o, ok := cur[r.Key()]; ok { c = &o } add(r.Kind, r.ID, r, c) } for i := range actual { o := &actual[i] if seen[o.Key()] || !cfg.Selected(o.Kind, o.ID) { continue } add(o.Kind, o.ID, nil, o) } sort.SliceStable(out, func(i, j int) bool { if out[i].Kind != out[j].Kind { return kindRank(out[i].Kind) < kindRank(out[j].Kind) } return out[i].ID < out[j].ID }) return out } // OpDelete is re-exported for the switch above (engine.Op is a string type). const OpDelete = engine.OpDelete func kindRank(k model.Kind) int { for i, x := range model.Kinds { if x == k { return i } } return len(model.Kinds) } // ClientPlan is the plan of one client restricted to a selection of objects. type ClientPlan struct { Client string `json:"client"` Plan engine.Plan `json:"plan"` Digest string `json:"digest"` SDN bool `json:"sdn"` Error string `json:"error,omitempty"` // client or reference unreadable } // Plans returns the plans of the given clients restricted to the given objects // (empty: everything), as a dry-run shows them and an apply checks them. func (st *State) Plans(keys []string) []ClientPlan { var out []ClientPlan for _, c := range st.Clients { cp := ClientPlan{Client: c.Name} switch { case st.Reference.Error != "": cp.Error = "reference: " + st.Reference.Error case c.Error != "": cp.Error = c.Error default: cp.Plan = engine.Filter(c.Plan, keys) cp.Digest = cp.Plan.Digest() cp.SDN = cp.Plan.HasSDN() } out = append(out, cp) } return out } // Apply writes one plan on a client, then checks that the selected objects are // in line with the reference (same reference objects as the plan). func (st *State) Apply(ctx context.Context, cp ClientPlan, keys []string, log func(engine.Step)) error { cl, ok := st.cfg.Client(cp.Client) if !ok { return fmt.Errorf("unknown client %q", cp.Client) } c, err := cl.PVE() if err != nil { return err } res := engine.ApplyWithLog(ctx, c, cp.Plan, log) if res.Err != nil { return res.Err } start := time.Now() err = engine.Verify(ctx, c, func(actual []model.Object) engine.Plan { return engine.Filter(engine.MakePlan(st.cfg, cl, st.ref, actual), keys) }, st.cfg.Kinds) s := engine.Step{At: start.UTC(), Action: "verify"} if err != nil { s.Error = err.Error() } log(s) return err } // Check tests the access to a cluster: versions (9.x only) and token privileges. type Check struct { Nodes []Node `json:"nodes,omitempty"` Error string `json:"error,omitempty"` Warnings []string `json:"warnings,omitempty"` } // CheckCluster runs the same checks as the "check" command. func CheckCluster(ctx context.Context, cl config.Cluster, write, dcOptions bool) Check { var res Check c, err := cl.PVE() if err != nil { res.Error = err.Error() return res } vs, err := c.CheckVersion(ctx) res.Nodes = nodes(vs) if err != nil { res.Error = err.Error() return res } perms, err := c.Permissions(ctx) if err != nil { res.Warnings = append(res.Warnings, fmt.Sprintf("token privileges unreadable (%v)", err)) return res } res.Warnings = PrivilegeWarnings(perms, write, dcOptions) return res } // PrivilegeWarnings flags a reference token able to write, and a client token // missing the privileges needed by the synchronized kinds. func PrivilegeWarnings(perms map[string]map[string]any, write, dcOptions bool) []string { has := func(path, priv string) bool { _, ok := perms[path][priv]; return ok } anyPath := func(priv string) bool { for p := range perms { if has(p, priv) { return true } } return false } var w []string if !write { for _, p := range []string{"Sys.Modify", "SDN.Allocate", "Permissions.Modify"} { if anyPath(p) { w = append(w, "le token de référence possède "+p+" : un token en lecture seule est attendu") } } return w } if !has("/sdn", "SDN.Allocate") && !has("/sdn/zones", "SDN.Allocate") { w = append(w, "SDN.Allocate absent sur /sdn : la synchro SDN échouera") } if dcOptions && !has("/", "Sys.Modify") { w = append(w, "Sys.Modify absent sur / : les options datacenter ne pourront pas être écrites") } if !has("/", "Sys.Audit") { w = append(w, "Sys.Audit absent sur / : options datacenter et versions illisibles") } return w } // nodes converts the versions read by the PVE client. func nodes(vs []pve.NodeVersion) []Node { out := make([]Node, 0, len(vs)) for _, v := range vs { out = append(out, Node{Name: v.Node, Version: v.Version}) } return out }