// Package engine implements the synchronization pipeline: // read source -> normalize -> diff -> plan -> apply -> verify. package engine import ( "context" "encoding/json" "fmt" "sort" "time" "pve-confsync/internal/adapter" "pve-confsync/internal/config" "pve-confsync/internal/model" "pve-confsync/internal/pve" ) // Read loads and normalizes the objects of the given kinds from a cluster. // The reference is read in ReadRunning mode (applied SDN configuration only), // clients in ReadCurrent mode (what the next writes will modify). func Read(ctx context.Context, c *pve.Client, kinds []model.Kind, mode adapter.ReadMode) (model.Snapshot, error) { snap := model.Snapshot{Cluster: c.Name(), ExportedAt: time.Now().UTC()} vs, err := c.CheckVersion(ctx) if err != nil { return snap, err } if len(vs) > 0 { snap.PVEVersion = vs[0].Version } want := map[model.Kind]bool{} for _, k := range kinds { want[k] = true } // subnets need the vnets, vnets need the zones: read the dependencies even // if their kind is not synchronized, but only keep the requested kinds. need := map[model.Kind]bool{} for k := range want { need[k] = true } if need[model.KindSubnet] { need[model.KindVNet] = true } if need[model.KindVNet] { need[model.KindZone] = true } if need[model.KindPassword] { need[model.KindUser] = true } var all []model.Object for _, a := range adapter.All() { if !need[a.Kind()] { continue } objs, warns, err := a.List(ctx, c, mode, all) if err != nil { return snap, fmt.Errorf("cluster %s: read %s: %w", c.Name(), a.Kind(), err) } snap.Warnings = append(snap.Warnings, warns...) all = append(all, objs...) } for _, o := range all { if want[o.Kind] { snap.Objects = append(snap.Objects, o) } } model.Sort(snap.Objects) return snap, nil } // Op is the action planned on one object. type Op string const ( OpCreate Op = "create" OpUpdate Op = "update" OpDelete Op = "delete" ) // Change is one planned action. Set holds the new values, Old the current ones. type Change struct { Kind model.Kind `json:"kind"` ID string `json:"id"` Op Op `json:"op"` Set map[string]any `json:"set,omitempty"` Unset []string `json:"unset,omitempty"` Old map[string]any `json:"old,omitempty"` Target model.Object `json:"-"` } // MarshalJSON hides the secret fields (password hashes): plans are shown in the // UI and kept in the journal, the real values only live in memory until applied. func (c Change) MarshalJSON() ([]byte, error) { type plain Change red := func(m map[string]any) map[string]any { if m == nil { return nil } out := make(map[string]any, len(m)) for k, v := range m { if adapter.IsSensitive(c.Kind, k) { v = adapter.Redact(v) } out[k] = v } return out } p := plain(c) p.Set, p.Old = red(c.Set), red(c.Old) return json.Marshal(p) } // Plan is the list of changes for one client, in execution order. type Plan struct { Client string `json:"client"` Changes []Change `json:"changes"` Errors []string `json:"errors,omitempty"` Warnings []string `json:"warnings,omitempty"` } // HasSDN tells whether the plan writes SDN objects (lock + reload needed). func (p Plan) HasSDN() bool { for _, c := range p.Changes { if a, _ := adapter.ByKind(c.Kind); a != nil && a.SDN() { return true } } return false } // Empty tells whether there is nothing to do. func (p Plan) Empty() bool { return len(p.Changes) == 0 } // MakePlan computes the changes that bring the client in line with the reference. func MakePlan(cfg *config.Config, cl *config.Client, ref, actual []model.Object) Plan { plan := Plan{Client: cl.Name} var deletes []Change for _, k := range cfg.Kinds { a, err := adapter.ByKind(k) if err != nil { plan.Errors = append(plan.Errors, err.Error()) continue } cur := map[string]model.Object{} for _, o := range actual { if o.Kind == k { cur[o.ID] = o } } inRef := map[string]bool{} for _, r := range ref { if r.Kind != k || !cfg.Selected(k, r.ID) { continue } inRef[r.ID] = true mapping := model.Normalize(cl.Mapping[k][r.ID]) a.Canonicalize(mapping) desired, managed := desiredFields(cfg, cl, a, r, mapping) o, exists := cur[r.ID] if !exists { if a.Singleton() { plan.Errors = append(plan.Errors, fmt.Sprintf("%s: not found on client", r.Key())) continue } for _, f := range a.Required(model.Object{Kind: k, ID: r.ID, Fields: desired}) { if _, ok := desired[f]; !ok { plan.Errors = append(plan.Errors, fmt.Sprintf("%s: field %q required for creation, add it to the mapping of client %s", r.Key(), f, cl.Name)) } } plan.Changes = append(plan.Changes, Change{Kind: k, ID: r.ID, Op: OpCreate, Set: desired, Target: model.Object{Kind: k, ID: r.ID, Fields: desired}}) continue } ch := Change{Kind: k, ID: r.ID, Op: OpUpdate, Set: map[string]any{}, Old: map[string]any{}, Target: o} for f, v := range desired { if !model.Equal(v, o.Fields[f]) { if a.Immutable(f) { plan.Errors = append(plan.Errors, fmt.Sprintf("%s: field %q differs (%s -> %s) and cannot be changed, re-creation needed", r.Key(), f, model.Format(o.Fields[f]), model.Format(v))) continue } ch.Set[f] = v if old, ok := o.Fields[f]; ok { ch.Old[f] = old } } } // fields only present on the client are managed too when their policy is "sync" for f := range o.Fields { if cfg.Policy(cl, a, f) == adapter.Sync { managed[f] = true } } for f := range managed { if _, want := desired[f]; want { continue } if old, has := o.Fields[f]; has { ch.Unset = append(ch.Unset, f) ch.Old[f] = old } } sort.Strings(ch.Unset) if len(ch.Set) > 0 || len(ch.Unset) > 0 { plan.Changes = append(plan.Changes, ch) } } if cfg.Prune && !a.Singleton() { for id, o := range cur { if !inRef[id] && cfg.Selected(k, id) { deletes = append(deletes, Change{Kind: k, ID: id, Op: OpDelete, Old: o.Fields, Target: o}) } } } } checkParents(&plan, cl.Name, actual) // deletions last, dependents first (subnets, vnets, zones) sort.SliceStable(deletes, func(i, j int) bool { if deletes[i].Kind != deletes[j].Kind { return kindRank(deletes[i].Kind) > kindRank(deletes[j].Kind) } return deletes[i].ID < deletes[j].ID }) plan.Changes = append(plan.Changes, deletes...) return plan } // checkParents blocks the objects whose parents (groups of an account, account // of a password) are neither on the target nor created by the plan. func checkParents(plan *Plan, client string, actual []model.Object) { have := map[string]bool{} for _, o := range actual { have[o.Key()] = true } for _, c := range plan.Changes { if c.Op == OpCreate { have[c.Key()] = true } } for _, c := range plan.Changes { if c.Op == OpDelete { continue } for _, p := range adapter.Parents(Desired(c)) { if !have[p] { plan.Errors = append(plan.Errors, fmt.Sprintf("%s: %s absent on target %s and not replicated", c.Key(), p, client)) } } } } // Desired is the object as it will be after the change (fields of the target // with the new values). func Desired(c Change) model.Object { f := map[string]any{} for k, v := range c.Target.Fields { f[k] = v } for _, k := range c.Unset { delete(f, k) } for k, v := range c.Set { f[k] = v } return model.Object{Kind: c.Kind, ID: c.ID, Fields: f} } func kindRank(k model.Kind) int { for i, x := range model.Kinds { if x == k { return i } } return len(model.Kinds) } // desiredFields returns the field values the client must have, and the set of // fields the tool manages on this object (absent from desired => to delete). func desiredFields(cfg *config.Config, cl *config.Client, a adapter.Adapter, r model.Object, mapping map[string]any) (map[string]any, map[string]bool) { desired := map[string]any{} managed := map[string]bool{} // a field is managed when its policy is "sync", whatever its presence for f, v := range r.Fields { if cfg.Policy(cl, a, f) == adapter.Sync { desired[f] = v } } for f := range cfg.Policies[a.Kind()] { if cfg.Policy(cl, a, f) == adapter.Sync { managed[f] = true } } for f := range cl.Policies[a.Kind()] { if cfg.Policy(cl, a, f) == adapter.Sync { managed[f] = true } } for f := range r.Fields { if cfg.Policy(cl, a, f) == adapter.Sync { managed[f] = true } } // the client mapping wins over the reference (except forbidden fields) for f, v := range mapping { if a.Forbidden(f) || cfg.Policy(cl, a, f) == adapter.Ignore { continue } desired[f] = v managed[f] = true } return desired, managed }