// 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" "strings" "sync" "time" "pve-confsync/internal/adapter" "pve-confsync/internal/config" "pve-confsync/internal/engine" "pve-confsync/internal/model" "pve-confsync/internal/pve" "pve-confsync/internal/sshx" ) // 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, Clients: []Client{}, 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 } if adapter.IsSensitive(k, f) { // password hashes: a digest only fd.Ref, fd.Mapping = adapter.Redact(fd.Ref), adapter.Redact(fd.Mapping) fd.Current, fd.Target = adapter.Redact(fd.Current), adapter.Redact(fd.Target) } 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, skip []model.Kind) []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.Skip(c.Plan, keys, skip) 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, skip []model.Kind, 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.Skip(engine.MakePlan(st.cfg, cl, st.ref, actual), keys, skip) }, st.cfg.Kinds) s := engine.Step{At: start.UTC(), Action: "verify"} if err != nil { s.Error = err.Error() } log(s) return err } // Check is the result of a connection test: API with the root credentials // (login, PVE 9.x on every node) and SSH as root on the node. type Check struct { OK bool `json:"ok"` Error string `json:"error,omitempty"` Warnings []string `json:"warnings,omitempty"` API bool `json:"api"` SSH bool `json:"ssh"` Version string `json:"version,omitempty"` Nodes []Node `json:"nodes,omitempty"` // set when something is seen for the first time and must be recorded TLSFingerprint string `json:"tlsFingerprint,omitempty"` SSHHostKey string `json:"-"` SSHFingerprint string `json:"sshFingerprint,omitempty"` SSHMismatch bool `json:"sshMismatch,omitempty"` // the node presents another key than the recorded one } // CheckCluster tests a cluster. pinTLS: the TLS mode pins the certificate and // none is recorded yet (trust on first use). trust accepts a new SSH host key. func CheckCluster(ctx context.Context, cl config.Cluster, pinTLS, trust bool) Check { var res Check fail := func(err error) Check { res.Error = err.Error() return res } if pinTLS && cl.Fingerprint == "" { fp, err := pve.PeerFingerprint(ctx, cl.URL) if err != nil { return fail(fmt.Errorf("TLS: %w", err)) } cl.Fingerprint, res.TLSFingerprint = fp, fp } if cl.Insecure { res.Warnings = append(res.Warnings, "TLS certificate not verified (lab only)") } key := cl.SSHHostKey cl.SSHHostKey = "" // API only for now c, err := cl.PVE() if err != nil { return fail(err) } if err := c.Login(ctx); err != nil { return fail(fmt.Errorf("API: %w", err)) } vs, err := c.CheckVersion(ctx) res.Nodes = nodes(vs) if len(vs) > 0 { res.Version = vs[0].Version } if err != nil { return fail(err) } res.API = true o := cl.SSHOptions() if key == "" || trust { seen, fp, err := sshx.Scan(ctx, o) if err != nil { return fail(fmt.Errorf("SSH: %w", err)) } key, res.SSHHostKey, res.SSHFingerprint = seen, seen, fp } o.HostKey = key sh, err := sshx.New(o) if err != nil { return fail(fmt.Errorf("SSH: %w", err)) } out, err := sh.Run(ctx, "pveversion", nil) if err != nil { if strings.Contains(err.Error(), "host key mismatch") { _, fp, _ := sshx.Scan(ctx, o) res.SSHMismatch, res.SSHFingerprint = true, fp return fail(fmt.Errorf("SSH: node %s presents host key %s instead of the recorded %s", o.Host, fp, sshx.Fingerprint(key))) } return fail(fmt.Errorf("SSH: %w", err)) } if !strings.HasPrefix(strings.TrimSpace(string(out)), "pve-manager/") { return fail(fmt.Errorf("SSH: %s is not a PVE node (pveversion: %q)", o.Host, strings.TrimSpace(string(out)))) } res.SSH, res.OK = true, true return res } // 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 }