Compare commits

..

5 Commits

Author SHA1 Message Date
c28620324a update go mod 2025-12-09 00:48:54 +00:00
d5fea5e7b8 reimplement synchronization for sync methods 2025-11-09 23:59:19 +00:00
79eecdf211 improve various log messages,
improve http return codes
2025-11-08 01:00:38 +00:00
f83a26ff6b update go mod 2025-10-10 21:55:08 +00:00
612f2a159f update go mod 2025-10-02 17:46:52 +00:00
4 changed files with 213 additions and 156 deletions

View File

@@ -25,7 +25,7 @@ func Run() {
flag.Parse()
config := GetConfig(*configPath)
log.Printf("Initialized config from %s", *configPath)
log.Printf("[INFO] initialized config from %s", *configPath)
token := fmt.Sprintf(`%s@%s!%s`, config.PVE.Token.USER, config.PVE.Token.REALM, config.PVE.Token.ID)
client = NewClient(config.PVE.URL, token, config.PVE.Token.Secret)
@@ -35,13 +35,13 @@ func Run() {
cluster := Cluster{}
cluster.Init(client)
start := time.Now()
log.Printf("Starting cluster sync\n")
log.Printf("[INFO] starting cluster sync\n")
cluster.Sync()
log.Printf("Synced cluster in %fs\n", time.Since(start).Seconds())
log.Printf("[INFO] synced cluster in %fs\n", time.Since(start).Seconds())
// set repeating update for full rebuilds
ticker := time.NewTicker(time.Duration(config.ReloadInterval) * time.Second)
log.Printf("Initialized cluster sync interval of %ds", config.ReloadInterval)
log.Printf("[INFO] initialized cluster sync interval of %ds", config.ReloadInterval)
channel := make(chan bool)
go func() {
for {
@@ -50,9 +50,9 @@ func Run() {
return
case <-ticker.C:
start := time.Now()
log.Printf("Starting cluster sync\n")
log.Printf("[INFO] starting cluster sync\n")
cluster.Sync()
log.Printf("Synced cluster in %fs\n", time.Since(start).Seconds())
log.Printf("[INFO] synced cluster in %fs\n", time.Since(start).Seconds())
}
}
}()
@@ -66,13 +66,24 @@ func Run() {
}
})
router.GET("/", func(c *gin.Context) {
v, err := cluster.Get()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
} else {
c.JSON(http.StatusOK, gin.H{"cluster": v})
return
}
})
router.GET("/nodes/:node", func(c *gin.Context) {
nodeid := c.Param("node")
node, err := cluster.GetNode(nodeid)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
} else {
c.JSON(http.StatusOK, gin.H{"node": node})
@@ -80,25 +91,11 @@ func Run() {
}
})
router.GET("/nodes/:node/devices", func(c *gin.Context) {
nodeid := c.Param("node")
node, err := cluster.GetNode(nodeid)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
} else {
c.JSON(http.StatusOK, gin.H{"devices": node.Devices})
return
}
})
router.GET("/nodes/:node/instances/:vmid", func(c *gin.Context) {
nodeid := c.Param("node")
vmid, err := strconv.ParseUint(c.Param("vmid"), 10, 64)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", c.Param("instance"))})
c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", c.Param("instance"))})
return
}
@@ -110,7 +107,7 @@ func Run() {
} else {
instance, err := node.GetInstance(uint(vmid))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
} else {
c.JSON(http.StatusOK, gin.H{"instance": instance})
@@ -122,9 +119,16 @@ func Run() {
router.POST("/sync", func(c *gin.Context) {
//go func() {
start := time.Now()
log.Printf("Starting cluster sync\n")
cluster.Sync()
log.Printf("Synced cluster in %fs\n", time.Since(start).Seconds())
log.Printf("[INFO] starting cluster sync\n")
err := cluster.Sync()
if err != nil {
log.Printf("[ERR ] failed to sync cluster: %s", err)
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
} else {
log.Printf("[INFO] synced cluster in %fs\n", time.Since(start).Seconds())
return
}
//}()
})
@@ -132,13 +136,14 @@ func Run() {
nodeid := c.Param("node")
//go func() {
start := time.Now()
log.Printf("Starting %s sync\n", nodeid)
err := cluster.RebuildHost(nodeid)
log.Printf("[INFO] starting %s sync\n", nodeid)
err := cluster.RebuildNode(nodeid)
if err != nil {
log.Printf("Failed to sync %s: %s", nodeid, err.Error())
log.Printf("[ERR ] failed to sync %s: %s", nodeid, err.Error())
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
} else {
log.Printf("Synced %s in %fs\n", nodeid, time.Since(start).Seconds())
log.Printf("[INFO] synced %s in %fs\n", nodeid, time.Since(start).Seconds())
return
}
//}()
@@ -154,30 +159,34 @@ func Run() {
//go func() {
start := time.Now()
log.Printf("Starting %s.%d sync\n", nodeid, vmid)
log.Printf("[INFO] starting %s.%d sync\n", nodeid, vmid)
node, err := cluster.GetNode(nodeid)
if err != nil {
log.Printf("Failed to sync %s.%d: %s", nodeid, vmid, err.Error())
log.Printf("[ERR ] failed to sync %s.%d: %s", nodeid, vmid, err.Error())
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
}
instance, err := node.GetInstance(uint(vmid))
if err != nil {
log.Printf("Failed to sync %s.%d: %s", nodeid, vmid, err.Error())
log.Printf("[ERR ] failed to sync %s.%d: %s", nodeid, vmid, err.Error())
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
}
err = node.RebuildInstance(instance.Type, uint(vmid))
if err != nil {
log.Printf("Failed to sync %s.%d: %s", nodeid, vmid, err.Error())
log.Printf("[ERR ] failed to sync %s.%d: %s", nodeid, vmid, err.Error())
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
} else {
log.Printf("Synced %s.%d in %fs\n", nodeid, vmid, time.Since(start).Seconds())
log.Printf("[INFO] synced %s.%d in %fs\n", nodeid, vmid, time.Since(start).Seconds())
return
}
//}()
})
log.Printf("[INFO] starting API listening on 0.0.0.0:%d", config.ListenPort)
router.Run("0.0.0.0:" + strconv.Itoa(config.ListenPort))
}

View File

@@ -10,7 +10,26 @@ func (cluster *Cluster) Init(pve ProxmoxClient) {
cluster.pve = pve
}
func (cluster *Cluster) Get() (*Cluster, error) {
cluster_ch := make(chan *Cluster)
err_ch := make(chan error)
go func() {
// aquire cluster lock
cluster.lock.Lock()
defer cluster.lock.Unlock()
cluster_ch <- cluster
err_ch <- nil
}()
return <-cluster_ch, <-err_ch
}
// hard sync cluster
func (cluster *Cluster) Sync() error {
err_ch := make(chan error)
go func() {
// aquire lock on cluster, release on return
cluster.lock.Lock()
defer cluster.lock.Unlock()
@@ -20,19 +39,23 @@ func (cluster *Cluster) Sync() error {
// get all nodes
nodes, err := cluster.pve.Nodes()
if err != nil {
return err
err_ch <- err
return
}
// for each node:
for _, hostName := range nodes {
// rebuild node
err := cluster.RebuildHost(hostName)
err := cluster.RebuildNode(hostName)
if err != nil { // if an error was encountered, continue and log the error
log.Print(err.Error())
continue
log.Printf("[ERR ] %s", err)
} else { // otherwise log success
log.Printf("[INFO] successfully synced node %s", hostName)
}
}
err_ch <- nil
}()
return nil
return <-err_ch
}
// get a node in the cluster
@@ -59,47 +82,59 @@ func (cluster *Cluster) GetNode(hostName string) (*Node, error) {
}
}()
host := <-host_ch
err := <-err_ch
return host, err
return <-host_ch, <-err_ch
}
func (cluster *Cluster) RebuildHost(hostName string) error {
// hard sync node
// returns error if the node could not be reached
func (cluster *Cluster) RebuildNode(hostName string) error {
err_ch := make(chan error)
go func() {
host, err := cluster.pve.Node(hostName)
if err != nil { // host is probably down or otherwise unreachable
return fmt.Errorf("error retrieving %s: %s, possibly down?", hostName, err.Error())
if err != nil && cluster.Nodes[hostName] == nil { // host is unreachable and did not exist previously
// return an error because we requested to sync a node that was not already in the cluster
err_ch <- fmt.Errorf("error retrieving %s: %s", hostName, err.Error())
}
// aquire lock on host, release on return
host.lock.Lock()
defer host.lock.Unlock()
if err != nil && cluster.Nodes[hostName] != nil { // host is unreachable and did exist previously
// assume the node is down or gone and delete from cluster
delete(cluster.Nodes, hostName)
err_ch <- nil
}
cluster.Nodes[hostName] = host
// get node's VMs
vms, err := host.VirtualMachines()
if err != nil {
return err
err_ch <- err
}
for _, vmid := range vms {
err := host.RebuildInstance(VM, vmid)
if err != nil { // if an error was encountered, continue and log the error
log.Print(err.Error())
continue
log.Printf("[ERR ] %s", err)
} else {
log.Printf("[INFO] successfully synced vm %s.%d", hostName, vmid)
}
}
// get node's CTs
cts, err := host.Containers()
if err != nil {
return err
err_ch <- err
}
for _, vmid := range cts {
err := host.RebuildInstance(CT, vmid)
if err != nil {
return err
if err != nil { // if an error was encountered, continue and log the error
log.Printf("[ERR ] %s", err)
} else {
log.Printf("[INFO] successfully synced ct %s.%d", hostName, vmid)
}
}
@@ -112,7 +147,9 @@ func (cluster *Cluster) RebuildHost(hostName string) error {
device.Reserved = reserved
}
return nil
err_ch <- nil
}()
return <-err_ch
}
func (host *Node) GetInstance(vmid uint) (*Instance, error) {
@@ -138,33 +175,41 @@ func (host *Node) GetInstance(vmid uint) (*Instance, error) {
}
}()
instance := <-instance_ch
err := <-err_ch
return instance, err
return <-instance_ch, <-err_ch
}
// hard sync instance
// returns error if the instance could not be reached
func (host *Node) RebuildInstance(instancetype InstanceType, vmid uint) error {
err_ch := make(chan error)
go func() {
instanceID := InstanceID(vmid)
var instance *Instance
var err error
if instancetype == VM {
var err error
instance, err = host.VirtualMachine(vmid)
if err != nil {
return fmt.Errorf("error retrieving %d: %s, possibly down?", vmid, err.Error())
}
} else if instancetype == CT {
var err error
instance, err = host.Container(vmid)
if err != nil {
return fmt.Errorf("error retrieving %d: %s, possibly down?", vmid, err.Error())
}
if err != nil && host.Instances[instanceID] == nil { // instance is unreachable and did not exist previously
// return an error because we requested to sync an instance that was not already in the cluster
err_ch <- fmt.Errorf("error retrieving %s.%d: %s", host.Name, instanceID, err.Error())
}
// aquire lock on instance, release on return
instance.lock.Lock()
defer instance.lock.Unlock()
host.Instances[InstanceID(vmid)] = instance
if err != nil && host.Instances[instanceID] != nil { // host is unreachable and did exist previously
// assume the instance is gone and delete from cluster
delete(host.Instances, instanceID)
err_ch <- nil
}
host.Instances[instanceID] = instance
for volid := range instance.configDisks {
instance.RebuildVolume(host, volid)
@@ -182,7 +227,10 @@ func (host *Node) RebuildInstance(instancetype InstanceType, vmid uint) error {
instance.RebuildBoot()
}
return nil
err_ch <- nil
}()
return <-err_ch
}
func (instance *Instance) RebuildVolume(host *Node, volid string) error {
@@ -215,10 +263,11 @@ func (instance *Instance) RebuildNet(netid string) error {
return nil
}
func (instance *Instance) RebuildDevice(host *Node, deviceid string) error {
func (instance *Instance) RebuildDevice(host *Node, deviceid string) {
instanceDevice, ok := instance.configHostPCIs[deviceid]
if !ok { // if device does not exist
return fmt.Errorf("%s not found in devices", deviceid)
log.Printf("[WARN] %s not found in devices", deviceid)
return
}
hostDeviceBusID := DeviceID(strings.Split(instanceDevice, ",")[0])
@@ -234,9 +283,6 @@ func (instance *Instance) RebuildDevice(host *Node, deviceid string) error {
}
instance.Devices[DeviceID(instanceDeviceBusID)].Device_ID = DeviceID(deviceid)
instance.Devices[DeviceID(instanceDeviceBusID)].Value = instanceDevice
return nil
}
func (instance *Instance) RebuildBoot() {
@@ -265,7 +311,7 @@ func (instance *Instance) RebuildBoot() {
instance.Boot.Enabled = append(instance.Boot.Enabled, val)
delete(eligibleBoot, bootTarget)
} else { // item is not eligible for boot but is included in the boot order
log.Printf("Encountered enabled but non-eligible boot target %s in instance %s\n", bootTarget, instance.Name)
log.Printf("[WARN] encountered enabled but non-eligible boot target %s in instance %s\n", bootTarget, instance.Name)
delete(eligibleBoot, bootTarget)
}
}
@@ -277,7 +323,7 @@ func (instance *Instance) RebuildBoot() {
} else if val, ok := instance.Nets[NetID(bootTarget)]; ok && isEligible { // if the item is eligible and is in nets
instance.Boot.Disabled = append(instance.Boot.Disabled, val)
} else { // item is not eligible and is not already in the boot order, skip adding to model
log.Printf("Encountered disabled and non-eligible boot target %s in instance %s\n", bootTarget, instance.Name)
log.Printf("[WARN] encountered disabled and non-eligible boot target %s in instance %s\n", bootTarget, instance.Name)
}
}
}

View File

@@ -9,7 +9,7 @@ import (
type Cluster struct {
lock sync.Mutex
pve ProxmoxClient
Nodes map[string]*Node
Nodes map[string]*Node `json:"nodes"`
}
type Node struct {
@@ -88,7 +88,6 @@ type Device struct {
Vendor_Name string `json:"vendor_name"`
Functions map[FunctionID]*Function `json:"functions"`
Reserved bool `json:"reserved"`
Value string
}
type FunctionID string

37
go.mod
View File

@@ -1,32 +1,33 @@
module proxmoxaas-fabric
go 1.25.1
go 1.25.5
require (
github.com/gin-gonic/gin v1.10.1
github.com/gin-gonic/gin v1.11.0
github.com/luthermonson/go-proxmox v0.2.3
)
require (
github.com/buger/goterm v1.0.4 // indirect
github.com/bytedance/gopkg v0.1.3 // indirect
github.com/bytedance/sonic v1.14.1 // indirect
github.com/bytedance/sonic/loader v0.3.0 // indirect
github.com/bytedance/sonic v1.14.2 // indirect
github.com/bytedance/sonic/loader v0.4.0 // indirect
github.com/cloudwego/base64x v0.1.6 // indirect
github.com/diskfs/go-diskfs v1.7.0 // indirect
github.com/djherbis/times v1.6.0 // indirect
github.com/elliotwutingfeng/asciiset v0.0.0-20250812055617-fb43ac3ba420 // indirect
github.com/gabriel-vasile/mimetype v1.4.10 // indirect
github.com/elliotwutingfeng/asciiset v0.0.0-20250912055424-93680c478db2 // indirect
github.com/gabriel-vasile/mimetype v1.4.11 // indirect
github.com/gin-contrib/sse v1.1.0 // indirect
github.com/go-playground/locales v0.14.1 // indirect
github.com/go-playground/universal-translator v0.18.1 // indirect
github.com/go-playground/validator/v10 v10.27.0 // indirect
github.com/go-playground/validator/v10 v10.28.0 // indirect
github.com/goccy/go-json v0.10.5 // indirect
github.com/goccy/go-yaml v1.19.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/jinzhu/copier v0.4.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/compress v1.18.0 // indirect
github.com/klauspost/compress v1.18.2 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
github.com/leodido/go-urn v1.4.0 // indirect
github.com/magefile/mage v1.15.0 // indirect
@@ -36,14 +37,16 @@ require (
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
github.com/pierrec/lz4/v4 v4.1.22 // indirect
github.com/pkg/xattr v0.4.12 // indirect
github.com/quic-go/qpack v0.6.0 // indirect
github.com/quic-go/quic-go v0.57.1 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.3.0 // indirect
github.com/ulikunitz/xz v0.5.12 // indirect
golang.org/x/arch v0.21.0 // indirect
golang.org/x/crypto v0.42.0 // indirect
golang.org/x/net v0.44.0 // indirect
golang.org/x/sys v0.36.0 // indirect
golang.org/x/text v0.29.0 // indirect
google.golang.org/protobuf v1.36.9 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
github.com/ugorji/go/codec v1.3.1 // indirect
github.com/ulikunitz/xz v0.5.15 // indirect
go.uber.org/mock v0.6.0 // indirect
golang.org/x/arch v0.23.0 // indirect
golang.org/x/crypto v0.46.0 // indirect
golang.org/x/net v0.48.0 // indirect
golang.org/x/sys v0.39.0 // indirect
golang.org/x/text v0.32.0 // indirect
google.golang.org/protobuf v1.36.10 // indirect
)