pve-confsync/internal/pve/cluster.go

package pve

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"time"
)

// SupportedMajor is the only Proxmox VE major release this tool accepts.
const SupportedMajor = 9

// NodeVersion is the pve-manager version of one node.
type NodeVersion struct {
	Node    string
	Version string // full pve-manager version, e.g. 9.1.4
	Release string // point release, e.g. 9.1
}

// Versions returns the version of every online node (GET /cluster/status, then
// GET /nodes/{node}/version). A standalone node without cluster falls back to GET /version.
func (c *Client) Versions(ctx context.Context) ([]NodeVersion, error) {
	var status []map[string]any
	if err := c.Get(ctx, "cluster/status", nil, &status); err != nil {
		return nil, err
	}
	var res []NodeVersion
	for _, s := range status {
		if Str(s["type"]) != "node" || Str(s["online"]) != "1" {
			continue
		}
		node := Str(s["name"])
		v, err := c.version(ctx, "nodes/"+url.PathEscape(node)+"/version")
		if err != nil {
			return nil, err
		}
		v.Node = node
		res = append(res, v)
	}
	if len(res) == 0 {
		v, err := c.version(ctx, "version")
		if err != nil {
			return nil, err
		}
		res = append(res, v)
	}
	return res, nil
}

func (c *Client) version(ctx context.Context, path string) (NodeVersion, error) {
	var v struct {
		Version string `json:"version"`
		Release string `json:"release"`
	}
	if err := c.Get(ctx, path, nil, &v); err != nil {
		return NodeVersion{}, err
	}
	return NodeVersion{Version: v.Version, Release: v.Release}, nil
}

// CheckVersion refuses any cluster with a node outside PVE 9.x.
func (c *Client) CheckVersion(ctx context.Context) ([]NodeVersion, error) {
	vs, err := c.Versions(ctx)
	if err != nil {
		return nil, fmt.Errorf("cluster %s: version check: %w", c.name, err)
	}
	for _, v := range vs {
		ref := v.Release
		if ref == "" {
			ref = v.Version
		}
		major, err := strconv.Atoi(strings.SplitN(ref, ".", 2)[0])
		if err != nil || major != SupportedMajor {
			return vs, fmt.Errorf("cluster %s: node %s runs PVE %q, only %d.x is supported",
				c.name, v.Node, ref, SupportedMajor)
		}
	}
	return vs, nil
}

// Permissions returns the effective privileges of the token (GET /access/permissions),
// as path -> privilege -> propagate flag.
func (c *Client) Permissions(ctx context.Context) (map[string]map[string]any, error) {
	var p map[string]map[string]any
	err := c.Get(ctx, "access/permissions", nil, &p)
	return p, err
}

// WaitTask polls a worker task until it stops and returns an error unless it ended "OK".
func (c *Client) WaitTask(ctx context.Context, upid string, poll time.Duration) error {
	node, err := upidNode(upid)
	if err != nil {
		return err
	}
	path := "nodes/" + url.PathEscape(node) + "/tasks/" + url.PathEscape(upid) + "/status"
	for {
		var st struct {
			Status     string `json:"status"`
			ExitStatus string `json:"exitstatus"`
		}
		if err := c.Get(ctx, path, nil, &st); err != nil {
			return err
		}
		if st.Status == "stopped" {
			if st.ExitStatus != "OK" {
				return fmt.Errorf("cluster %s: task %s ended with %q", c.name, upid, st.ExitStatus)
			}
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(poll):
		}
	}
}

// upidNode extracts the node from "UPID:<node>:<pid>:<pstart>:<starttime>:<type>:<id>:<user>:".
func upidNode(upid string) (string, error) {
	parts := strings.Split(upid, ":")
	if len(parts) < 3 || parts[0] != "UPID" || parts[1] == "" {
		return "", fmt.Errorf("invalid UPID %q", upid)
	}
	return parts[1], nil
}

// Str renders a decoded JSON scalar as a string (json.Number, bool as 1/0).
func Str(v any) string {
	switch t := v.(type) {
	case nil:
		return ""
	case string:
		return t
	case json.Number:
		return t.String()
	case bool:
		if t {
			return "1"
		}
		return "0"
	case float64:
		return strconv.FormatFloat(t, 'f', -1, 64)
	case int:
		return strconv.Itoa(t)
	default:
		return fmt.Sprint(t)
	}
}