Phase 0 : configuration YAML et moteur (lecture, plan, apply sous verrou SDN, vérification)

internal/config/config.go

@@ -0,0 +1,227 @@
+// Package config loads the YAML configuration of the CLI (phase 0): clusters,
+// field policies, per client mappings and object selection.
+// Token secrets are never written in this file: they are read from an
+// environment variable or a file (docker secret), see README.
+package config
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path"
+	"strings"
+	"time"
+
+	"gopkg.in/yaml.v3"
+
+	"pve-confsync/internal/adapter"
+	"pve-confsync/internal/model"
+	"pve-confsync/internal/pve"
+)
+
+// Cluster is the connection to one PVE cluster.
+type Cluster struct {
+	Name            string `yaml:"name"`
+	URL             string `yaml:"url"`
+	TokenID         string `yaml:"token_id"`
+	TokenSecretEnv  string `yaml:"token_secret_env,omitempty"`
+	TokenSecretFile string `yaml:"token_secret_file,omitempty"`
+	CAFile          string `yaml:"ca_file,omitempty"`
+	Fingerprint     string `yaml:"fingerprint,omitempty"`
+	Insecure        bool   `yaml:"insecure,omitempty"`
+	Timeout         string `yaml:"timeout,omitempty"`
+}
+
+// Client is a target cluster with its mapping (kind -> id -> field -> value)
+// and its own policy overrides.
+type Client struct {
+	Cluster  `yaml:",inline"`
+	Mapping  map[model.Kind]map[string]map[string]any `yaml:"mapping,omitempty"`
+	Policies map[model.Kind]map[string]adapter.Policy `yaml:"policies,omitempty"`
+}
+
+// Selection filters the objects of a kind with glob patterns on their id.
+type Selection struct {
+	Include []string `yaml:"include,omitempty"`
+	Exclude []string `yaml:"exclude,omitempty"`
+}
+
+// Config is the whole file.
+type Config struct {
+	Reference Cluster                                  `yaml:"reference"`
+	Clients   []Client                                 `yaml:"clients"`
+	Kinds     []model.Kind                             `yaml:"kinds,omitempty"`
+	Objects   map[model.Kind]Selection                 `yaml:"objects,omitempty"`
+	Policies  map[model.Kind]map[string]adapter.Policy `yaml:"policies,omitempty"`
+	Prune     bool                                     `yaml:"prune,omitempty"`
+}
+
+// Load reads and validates a configuration file.
+func Load(file string) (*Config, error) {
+	raw, err := os.ReadFile(file)
+	if err != nil {
+		return nil, err
+	}
+	var c Config
+	dec := yaml.NewDecoder(strings.NewReader(string(raw)))
+	dec.KnownFields(true) // typos must fail, not be silently ignored
+	if err := dec.Decode(&c); err != nil {
+		return nil, fmt.Errorf("%s: %w", file, err)
+	}
+	if len(c.Kinds) == 0 {
+		c.Kinds = append(c.Kinds, model.Kinds...)
+	}
+	return &c, c.validate()
+}
+
+func (c *Config) validate() error {
+	var errs []error
+	seen := map[string]bool{}
+	all := append([]Cluster{c.Reference}, clusters(c.Clients)...)
+	for i, cl := range all {
+		if cl.Name == "" {
+			errs = append(errs, fmt.Errorf("cluster #%d: name is required", i))
+		}
+		if seen[cl.Name] {
+			errs = append(errs, fmt.Errorf("cluster %s: duplicate name", cl.Name))
+		}
+		seen[cl.Name] = true
+		if cl.TokenSecretEnv == "" && cl.TokenSecretFile == "" {
+			errs = append(errs, fmt.Errorf("cluster %s: token_secret_env or token_secret_file is required", cl.Name))
+		}
+	}
+	for _, k := range c.Kinds {
+		if _, err := adapter.ByKind(k); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	check := func(where string, p map[model.Kind]map[string]adapter.Policy) {
+		for k, fields := range p {
+			a, err := adapter.ByKind(k)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("%s: %w", where, err))
+				continue
+			}
+			for f, pol := range fields {
+				switch pol {
+				case adapter.Sync, adapter.Ignore, adapter.Override:
+				default:
+					errs = append(errs, fmt.Errorf("%s: %s.%s: unknown policy %q", where, k, f, pol))
+				}
+				if pol == adapter.Sync && a.Forbidden(f) {
+					errs = append(errs, fmt.Errorf("%s: %s.%s cannot be synchronized", where, k, f))
+				}
+			}
+		}
+	}
+	check("policies", c.Policies)
+	for _, cl := range c.Clients {
+		check("client "+cl.Name, cl.Policies)
+		for k := range cl.Mapping {
+			if _, err := adapter.ByKind(k); err != nil {
+				errs = append(errs, fmt.Errorf("client %s mapping: %w", cl.Name, err))
+			}
+		}
+	}
+	return errors.Join(errs...)
+}
+
+func clusters(cs []Client) []Cluster {
+	out := make([]Cluster, len(cs))
+	for i, c := range cs {
+		out[i] = c.Cluster
+	}
+	return out
+}
+
+// Client returns the client with this name.
+func (c *Config) Client(name string) (*Client, bool) {
+	for i := range c.Clients {
+		if c.Clients[i].Name == name {
+			return &c.Clients[i], true
+		}
+	}
+	return nil, false
+}
+
+// PVE builds the API client of a cluster, resolving its token secret.
+func (cl Cluster) PVE() (*pve.Client, error) {
+	secret, err := cl.secret()
+	if err != nil {
+		return nil, err
+	}
+	var timeout time.Duration
+	if cl.Timeout != "" {
+		if timeout, err = time.ParseDuration(cl.Timeout); err != nil {
+			return nil, fmt.Errorf("cluster %s: timeout: %w", cl.Name, err)
+		}
+	}
+	return pve.New(pve.Options{
+		Name: cl.Name, BaseURL: cl.URL, TokenID: cl.TokenID, TokenSecret: secret,
+		CAFile: cl.CAFile, Fingerprint: cl.Fingerprint, Insecure: cl.Insecure, Timeout: timeout,
+	})
+}
+
+func (cl Cluster) secret() (string, error) {
+	if cl.TokenSecretEnv != "" {
+		if v := os.Getenv(cl.TokenSecretEnv); v != "" {
+			return v, nil
+		}
+		if cl.TokenSecretFile == "" {
+			return "", fmt.Errorf("cluster %s: environment variable %s is empty", cl.Name, cl.TokenSecretEnv)
+		}
+	}
+	b, err := os.ReadFile(cl.TokenSecretFile)
+	if err != nil {
+		return "", fmt.Errorf("cluster %s: token secret: %w", cl.Name, err)
+	}
+	return strings.TrimSpace(string(b)), nil
+}
+
+// Policy resolves the policy of a field for a client: client policy, then
+// global policy, then the adapter default. Forbidden fields are always ignored.
+func (c *Config) Policy(cl *Client, a adapter.Adapter, field string) adapter.Policy {
+	if a.Forbidden(field) {
+		return adapter.Ignore
+	}
+	if cl != nil {
+		if p, ok := cl.Policies[a.Kind()][field]; ok {
+			return p
+		}
+	}
+	if p, ok := c.Policies[a.Kind()][field]; ok {
+		return p
+	}
+	return a.DefaultPolicy(field)
+}
+
+// Selected tells whether an object id is part of the synchronization.
+func (c *Config) Selected(k model.Kind, id string) bool {
+	s, ok := c.Objects[k]
+	if !ok {
+		return true
+	}
+	in := len(s.Include) == 0
+	for _, p := range s.Include {
+		if m, _ := path.Match(p, id); m {
+			in = true
+			break
+		}
+	}
+	for _, p := range s.Exclude {
+		if m, _ := path.Match(p, id); m {
+			return false
+		}
+	}
+	return in
+}
+
+// HasKind tells whether a kind is synchronized.
+func (c *Config) HasKind(k model.Kind) bool {
+	for _, x := range c.Kinds {
+		if x == k {
+			return true
+		}
+	}
+	return false
+}

internal/engine/apply.go

@@ -0,0 +1,165 @@
+package engine
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"time"
+
+	"pve-confsync/internal/adapter"
+	"pve-confsync/internal/model"
+	"pve-confsync/internal/pve"
+)
+
+// Step is the outcome of one API action, for the execution journal.
+type Step struct {
+	At     time.Time `json:"at"`
+	Action string    `json:"action"`
+	Error  string    `json:"error,omitempty"`
+}
+
+// Result of an apply on one client.
+type Result struct {
+	Client string `json:"client"`
+	Steps  []Step `json:"steps"`
+	Err    error  `json:"-"`
+}
+
+func (r *Result) log(action string, err error) {
+	s := Step{At: time.Now().UTC(), Action: action}
+	if err != nil {
+		s.Error = err.Error()
+	}
+	r.Steps = append(r.Steps, s)
+}
+
+// TaskPoll is the poll interval of the SDN reload task.
+var TaskPoll = 2 * time.Second
+
+// Apply executes a plan. Non SDN changes are applied directly; SDN changes are
+// written under the global SDN lock (pve-network >= 1.1.3, PVE 9.0), then
+// committed with PUT /cluster/sdn. Any failure before the commit rolls the
+// pending SDN changes back (POST /cluster/sdn/rollback).
+func Apply(ctx context.Context, c *pve.Client, plan Plan) Result {
+	res := Result{Client: plan.Client}
+	if len(plan.Errors) > 0 {
+		res.Err = fmt.Errorf("plan has %d error(s), nothing applied", len(plan.Errors))
+		return res
+	}
+	if _, err := c.CheckVersion(ctx); err != nil {
+		res.Err = err
+		res.log("version check", err)
+		return res
+	}
+	var sdn []Change
+	for _, ch := range plan.Changes {
+		a, _ := adapter.ByKind(ch.Kind)
+		if a.SDN() {
+			sdn = append(sdn, ch)
+			continue
+		}
+		err := run(ctx, c, a, ch, "")
+		res.log(describe(ch), err)
+		if err != nil {
+			res.Err = err
+			return res
+		}
+	}
+	if len(sdn) == 0 {
+		return res
+	}
+
+	var token string
+	err := c.Post(ctx, "cluster/sdn/lock", nil, &token)
+	res.log("sdn lock", err)
+	if err != nil {
+		res.Err = fmt.Errorf("SDN lock (pending changes on the client?): %w", err)
+		return res
+	}
+	rollback := func(cause error) {
+		f := url.Values{"lock-token": {token}, "release-lock": {"1"}}
+		rerr := c.Post(ctx, "cluster/sdn/rollback", f, nil)
+		res.log("sdn rollback", rerr)
+		res.Err = cause
+		if rerr != nil {
+			res.Err = errors.Join(cause, fmt.Errorf("rollback failed, lock still held: %w", rerr))
+		}
+	}
+	for _, ch := range sdn {
+		a, _ := adapter.ByKind(ch.Kind)
+		err := run(ctx, c, a, ch, token)
+		res.log(describe(ch), err)
+		if err != nil {
+			rollback(err)
+			return res
+		}
+	}
+	var upid string
+	err = c.Put(ctx, "cluster/sdn", url.Values{"lock-token": {token}, "release-lock": {"1"}}, &upid)
+	res.log("sdn apply", err)
+	if err != nil {
+		rollback(err)
+		return res
+	}
+	// The commit is done: from here a failure is in the network reload of the
+	// nodes, the running configuration is already the new one.
+	err = c.WaitTask(ctx, upid, TaskPoll)
+	res.log("sdn reload task "+upid, err)
+	if err != nil {
+		res.Err = fmt.Errorf("SDN configuration committed but reload failed: %w", err)
+	}
+	return res
+}
+
+func run(ctx context.Context, c *pve.Client, a adapter.Adapter, ch Change, lock string) error {
+	switch ch.Op {
+	case OpCreate:
+		return a.Create(ctx, c, model.Object{Kind: ch.Kind, ID: ch.ID, Fields: ch.Set}, lock)
+	case OpUpdate:
+		return a.Update(ctx, c, ch.Target, ch.Set, ch.Unset, lock)
+	case OpDelete:
+		return a.Delete(ctx, c, ch.Target, lock)
+	}
+	return fmt.Errorf("unknown op %q", ch.Op)
+}
+
+func describe(ch Change) string {
+	return fmt.Sprintf("%s %s/%s", ch.Op, ch.Kind, ch.ID)
+}
+
+// Verify re-reads the client and checks that nothing is left to do and that
+// no node reports an SDN zone in error (GET /nodes/{node}/sdn/zones).
+func Verify(ctx context.Context, c *pve.Client, plan func(actual []model.Object) Plan, kinds []model.Kind) error {
+	snap, err := Read(ctx, c, kinds, adapter.ReadCurrent)
+	if err != nil {
+		return err
+	}
+	var errs []error
+	if p := plan(snap.Objects); !p.Empty() {
+		for _, ch := range p.Changes {
+			errs = append(errs, fmt.Errorf("still different after apply: %s", describe(ch)))
+		}
+	}
+	var status []map[string]any
+	if err := c.Get(ctx, "cluster/status", nil, &status); err != nil {
+		return errors.Join(append(errs, err)...)
+	}
+	for _, s := range status {
+		if pve.Str(s["type"]) != "node" || pve.Str(s["online"]) != "1" {
+			continue
+		}
+		node := pve.Str(s["name"])
+		var zones []map[string]any
+		if err := c.Get(ctx, "nodes/"+url.PathEscape(node)+"/sdn/zones", nil, &zones); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		for _, z := range zones {
+			if pve.Str(z["status"]) == "error" {
+				errs = append(errs, fmt.Errorf("node %s: SDN zone %s in error", node, pve.Str(z["zone"])))
+			}
+		}
+	}
+	return errors.Join(errs...)
+}

internal/engine/engine.go

@@ -0,0 +1,245 @@
+// Package engine implements the synchronization pipeline:
+// read source -> normalize -> diff -> plan -> apply -> verify.
+package engine
+
+import (
+	"context"
+	"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
+	}
+	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" yaml:"kind"`
+	ID     string         `json:"id" yaml:"id"`
+	Op     Op             `json:"op" yaml:"op"`
+	Set    map[string]any `json:"set,omitempty" yaml:"set,omitempty"`
+	Unset  []string       `json:"unset,omitempty" yaml:"unset,omitempty"`
+	Old    map[string]any `json:"old,omitempty" yaml:"old,omitempty"`
+	Target model.Object   `json:"-" yaml:"-"`
+}
+
+// Plan is the list of changes for one client, in execution order.
+type Plan struct {
+	Client   string   `json:"client" yaml:"client"`
+	Changes  []Change `json:"changes" yaml:"changes"`
+	Errors   []string `json:"errors,omitempty" yaml:"errors,omitempty"`
+	Warnings []string `json:"warnings,omitempty" yaml:"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})
+				}
+			}
+		}
+	}
+	// 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
+}
+
+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
+}