pve-confsync/internal/service/service.go

// 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
}