package nodesync import ( "context" "errors" "fmt" "sort" "sync" "time" "pve-confsync/internal/model" ) // NodeInfo is what was read from one node. type NodeInfo struct { Key string `json:"name"` // "cluster/node", "name" like the cluster cards Cluster string `json:"cluster"` Node string `json:"node"` Version string `json:"version,omitempty"` 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 interface or of the DNS servers. type Field struct { Name string `json:"name"` Policy Policy `json:"policy"` Ref any `json:"ref,omitempty"` // master Mapping any `json:"mapping,omitempty"` // value of the target mapping Current any `json:"current,omitempty"` // target Target any `json:"target,omitempty"` // after the replication Change string `json:"change,omitempty"` // add, set, unset or empty } // Row statuses (same as the cluster rows). const ( StatusSame = "same" StatusCreate = "create" StatusUpdate = "update" StatusDelete = "delete" StatusExtra = "extra" // only on the target, kept StatusError = "error" ) // Row is one object of a target: an interface, the DNS servers or a file. type Row struct { Kind string `json:"kind"` // item id ID string `json:"id"` Key string `json:"key"` Status string `json:"status"` Fields []Field `json:"fields,omitempty"` Diff []Line `json:"diff,omitempty"` Errors []string `json:"errors,omitempty"` } // Target is the state of one target node. type Target struct { NodeInfo Plan Plan `json:"plan"` Digest string `json:"digest"` Rows []Row `json:"rows"` Counts map[string]int `json:"counts"` } // State is a full comparison of the master with its targets. type State struct { At time.Time `json:"at"` Profile string `json:"profile"` Items []string `json:"items"` Prune bool `json:"prune"` Master NodeInfo `json:"reference"` // "reference" like the cluster state Targets []Target `json:"clients"` prof *Profile master Snapshot targets map[string]Snapshot eps map[string]Endpoint } // Analyze reads the master and the targets (in parallel) and compares them. // only limits the targets (empty: all). A node that cannot be read is // reported in its Error field, it does not stop the others. func Analyze(ctx context.Context, prof *Profile, eps map[string]Endpoint, only []string) *State { st := &State{At: time.Now().UTC(), Profile: prof.Name, Items: prof.Items, Prune: prof.Prune, Targets: []Target{}, prof: prof, targets: map[string]Snapshot{}, eps: eps} st.Master, st.master = read(ctx, eps, prof.Master, prof.Items) want := map[string]bool{} for _, n := range only { want[n] = true } var wg sync.WaitGroup var mu sync.Mutex for _, key := range prof.Targets { if len(only) > 0 && !want[key] { continue } st.Targets = append(st.Targets, Target{NodeInfo: info(key)}) idx := len(st.Targets) - 1 wg.Add(1) go func() { defer wg.Done() ni, snap := read(ctx, eps, key, prof.Items) mu.Lock() defer mu.Unlock() st.Targets[idx].NodeInfo = ni if ni.Error == "" { st.targets[key] = snap } }() } wg.Wait() for i := range st.Targets { t := &st.Targets[i] t.Counts = map[string]int{} if t.Error != "" || st.Master.Error != "" { continue } t.Plan = MakePlan(prof, st.master, st.targets[t.Key]) t.Digest = t.Plan.Digest() t.Rows = rows(prof, st.master, st.targets[t.Key], t.Plan) for _, r := range t.Rows { t.Counts[r.Status]++ } } return st } func info(key string) NodeInfo { ni := NodeInfo{Key: key} for i := len(key) - 1; i >= 0; i-- { if key[i] == '/' { ni.Cluster, ni.Node = key[:i], key[i+1:] break } } return ni } func read(ctx context.Context, eps map[string]Endpoint, key string, items []string) (NodeInfo, Snapshot) { ni := info(key) ni.ReadAt = time.Now().UTC() start := time.Now() ep, ok := eps[key] if !ok { ni.Error = fmt.Sprintf("unknown node %s (discover the nodes of its cluster)", key) return ni, Snapshot{} } snap, err := Read(ctx, ep, items) if err != nil { ni.Error = err.Error() } ni.Version = snap.Version ni.Warnings = snap.Warnings ni.Millis = time.Since(start).Milliseconds() return ni, snap } // rows describes every object of the master and of the target. func rows(prof *Profile, master, target Snapshot, plan Plan) []Row { changes := map[string]Change{} for _, c := range plan.Changes { changes[c.Key()] = c } errs := map[string][]string{} for _, e := range plan.Errors { if e.Key != "" { errs[e.Key] = append(errs[e.Key], e.Message) } } tm := prof.Mapping[target.Key] var out []Row seen := map[string]bool{} status := func(key string, inMaster bool) string { ch, has := changes[key] switch { case len(errs[key]) > 0: return StatusError case has: return string(ch.Op) case !inMaster: return StatusExtra } return StatusSame } addAPI := func(item, id string, ref, cur map[string]any) { key := item + "/" + id seen[key] = true ch := changes[key] r := Row{Kind: item, ID: id, Key: key, Status: status(key, ref != nil), Errors: errs[key]} names := map[string]bool{} for _, m := range []map[string]any{ref, cur, ch.Set} { for f := range m { names[f] = true } } for f := range tm.Values[key] { 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: prof.Policy(item, f), Ref: ref[f], Current: cur[f]} if v, ok := tm.Values[key][f]; ok { fd.Mapping = v } switch { case ch.Op == OpDelete || unset[f]: fd.Change = "unset" case ch.Set[f] != nil: fd.Target = ch.Set[f] fd.Change = "set" if fd.Current == nil { fd.Change = "add" } case fd.Current != nil: fd.Target = fd.Current } r.Fields = append(r.Fields, fd) } out = append(out, r) } for _, id := range prof.Items { it, _ := ItemByID(id) switch it.Kind { case KindNetwork: cur := map[string]model.Object{} for _, o := range target.Network { cur[o.ID] = o } for _, m := range master.Network { tid := rename(m.ID, tm.NICs) addAPI("network", tid, m.Fields, cur[tid].Fields) } for _, o := range target.Network { if !seen["network/"+o.ID] { addAPI("network", o.ID, nil, o.Fields) } } case KindDNS: addAPI("dns", "dns", master.DNS, target.DNS) case KindFile: paths := map[string]bool{} for p := range master.Files[id] { paths[p] = true } for p := range target.Files[id] { paths[p] = true } if len(master.Files[id]) == 0 && !it.Glob { continue } for p := range paths { key := id + "/" + p _, inMaster := master.Files[id][p] r := Row{Kind: id, ID: p, Key: key, Status: status(key, inMaster), Errors: errs[key], Diff: changes[key].Diff} out = append(out, r) } } } sort.SliceStable(out, func(i, j int) bool { if out[i].Kind != out[j].Kind { return itemRank(out[i].Kind) < itemRank(out[j].Kind) } return out[i].ID < out[j].ID }) return out } // TargetPlan is the plan of one target restricted to a selection of objects. type TargetPlan struct { Client string `json:"client"` // "client" like the cluster plans Plan Plan `json:"plan"` Digest string `json:"digest"` Network bool `json:"network"` Reloads []string `json:"reloads,omitempty"` Error string `json:"error,omitempty"` } // Plans returns the plans of the targets restricted to the given objects // (empty: everything), as a dry-run shows them and an apply checks them. func (st *State) Plans(keys []string) []TargetPlan { var out []TargetPlan for _, t := range st.Targets { tp := TargetPlan{Client: t.Key} switch { case st.Master.Error != "": tp.Error = "master: " + st.Master.Error case t.Error != "": tp.Error = t.Error default: tp.Plan = Filter(t.Plan, keys) tp.Digest = tp.Plan.Digest() tp.Network = tp.Plan.Network() tp.Reloads = tp.Plan.Reloads() } out = append(out, tp) } return out } // Apply writes one plan on its target, then reads the target again and checks // that the selected objects are in line with the master. func (st *State) Apply(ctx context.Context, tp TargetPlan, keys []string, log func(Step)) error { ep, ok := st.eps[tp.Client] if !ok { return fmt.Errorf("unknown node %q", tp.Client) } if err := Apply(ctx, ep, tp.Plan, log); err != nil { return err } start := time.Now() err := st.verify(ctx, ep, keys) s := Step{At: start.UTC(), Action: "verify"} if err != nil { s.Error = err.Error() } log(s) return err } func (st *State) verify(ctx context.Context, ep Endpoint, keys []string) error { snap, err := Read(ctx, ep, st.prof.Items) if err != nil { return err } p := Filter(MakePlan(st.prof, st.master, snap), keys) var errs []error for _, c := range p.Changes { errs = append(errs, fmt.Errorf("still different after apply: %s", c.Key())) } return errors.Join(errs...) }