Compare commits

...
65 Commits
Author SHA1 Message Date
alu cc9da88b88 update README 2026-10-27 21:11:54 +00:00
alu 6ebe27560e remove replace for common-lib
tronnet/workflows: Static Analysis -- Golang / static-go (push) Successful in 43s
2026-10-26 23:14:56 +00:00
alu 072691569e use logging mddleware from common-lib
tronnet/workflows: Static Analysis -- Golang / static-go (push) Successful in 16s
2026-10-26 18:29:08 +00:00
alu 612fbddb33 use json helper from common-lib HTTP helper
tronnet/workflows: Static Analysis -- Golang / static-go (push) Successful in 39s
2026-10-26 18:24:54 +00:00
alu 590486f153 migrate to net/http, update go mod
tronnet/workflows: Static Analysis -- Golang / static-go (push) Successful in 31s
2026-10-23 21:32:40 +00:00
alu b9538a6337 update go 1.27 2026-08-25 18:56:22 +00:00
alu 7bda820cac update common lib 2026-08-19 21:11:23 +00:00
alu ee404fbf89 update go 1.26.6 2026-08-17 19:49:01 +00:00
alu b630b16ec9 fix makefile phony targets 2026-08-03 18:24:58 +00:00
alu 7a48c47b4a update common lib 2026-07-24 22:11:15 +00:00
alu cd0ecf1a90 update go mod, remove unused make test target 2026-07-17 21:36:56 +00:00
alu f3defe4157 resolve minor race conditions 2026-07-10 21:29:51 +00:00
alu 027ed2dce6 improved synchronization mechanism using rwmutexes 2026-07-10 20:39:58 +00:00
alu 33f1adb326 comment MutexWithCheck 2026-07-07 23:37:37 +00:00
alu d1395cd7ec update to go 1.26.5 2026-07-07 23:15:55 +00:00
alu e0c06582ab new mutex model 2026-07-07 23:13:18 +00:00
alu 9767c50f77 improve sync and build interfaces 2026-07-07 19:34:03 +00:00
alu 54e345a308 fix possible erroneous return if sync and fetch occur simultaneously 2026-07-07 03:37:48 +00:00
alu a7b932f600 add Readme 2026-07-06 23:57:47 +00:00
alu 05e9f017db fix missing dist dir make issue 2026-07-06 20:42:54 +00:00
alu c449618012 fix multiple possible concurrent map write issues,
fix issue with VM instacne types returning prematurely
2026-07-02 18:51:31 +00:00
alu b506aae260 fix missing pool membership clearing logic in ResolvePoolMembership 2026-07-01 18:08:21 +00:00
alu cbda49ff69 move unit related constant to common lib 2026-06-26 20:58:06 +00:00
alu e2de9c5083 add enforcement for single instance single pool 2026-06-26 18:20:52 +00:00
alu ab6a03be3b cleanup syn logic in app 2026-06-25 21:27:38 +00:00
alu 759b38e68d update go mod 2026-06-17 16:41:55 +00:00
alu abbce2322b add cluster ok value and checks 2026-06-15 16:15:31 +00:00
alu bc849ed7ca fix possible deadlock on resolve pool membership error 2026-06-09 21:21:28 +00:00
alu 95e420aaaa update common lib 2026-06-08 18:54:49 +00:00
alu 952b76c555 fix deadlock on failed cluster sync 2026-06-06 23:22:46 +00:00
alu 1b884008c5 update go mod 2026-06-03 17:04:06 +00:00
alu ba3ea746aa reduce repeated work while getting volume info 2026-06-02 23:26:04 +00:00
alu 7f9ad821b7 cleanup go mod 2026-06-02 20:38:35 +00:00
alu f1deacee10 improve sync latency significantly by using goroutines 2026-06-01 21:17:51 +00:00
alu a98745dd17 update to go-proxmox 0.7.0 2026-06-01 18:24:45 +00:00
alu 34bf10dfe3 update ResolvePoolMembership to be go routine 2026-05-29 22:16:35 +00:00
alu 606277e127 remove unused cluster pointer from host 2026-05-29 22:02:50 +00:00
alu a9d55673c7 update go mod, fix possible typecasting issues 2026-05-28 20:18:55 +00:00
alu 01c1d3cf60 update go mod 2026-05-28 20:03:29 +00:00
alu 64a018f1b5 remove tls skip verify, use os.Root for config reading 2026-05-28 19:51:41 +00:00
alu f430589512 fix issues with pool membership synchronization 2026-05-27 22:21:59 +00:00
alu 72ca7b8fc5 add missing error handling 2026-05-27 18:59:45 +00:00
alu dd54866c4e minor improvements, update common lib 2026-05-26 20:24:37 +00:00
alu 326c392d3d fix desync issue with instance pool membership 2026-05-23 17:37:49 +00:00
alu 4ce059be22 update common lib 2026-05-22 22:35:56 +00:00
alu ba8ba596b1 add pool to instance mapping to sync 2026-05-22 18:49:34 +00:00
alu 5474435dee update proxmoxaas-common-lib 2026-04-27 21:57:50 +00:00
alu 6c511a0291 Update configs/template.config.json 2026-04-23 22:35:45 +00:00
alu 0baf64ebf7 update go to 1.26.2, update proxmoxaas-common-lib 2026-04-22 16:29:52 +00:00
alu eb2f70f455 update go mod 2026-04-19 18:12:06 +00:00
alu fc30afa4f1 fix spelling 2026-04-19 17:22:40 +00:00
alu 8230092da3 add comment to types with importing proxmoxaas-common-lib types 2026-04-19 01:01:25 +00:00
alu e0a8f6a231 fix submodule and go linking issue 2026-04-19 00:54:17 +00:00
alu af380868b2 temp fix 2026-04-19 00:44:37 +00:00
alu 3a759e0583 add and switch type usage to common library types 2026-04-19 00:37:35 +00:00
alu 99bb5f0691 minor refactor of config values, update go mod 2026-04-04 22:27:11 +00:00
alu b74696f566 update go mod 2026-03-25 17:41:56 +00:00
alu daa88ac0d3 fix error message 2026-02-26 06:03:17 +00:00
alu fa6db0c445 add error message on app exit, update go mod 2026-02-26 06:02:18 +00:00
alu 354fa70856 update go mod 2026-02-12 06:00:16 +00:00
alu c28620324a update go mod 2025-12-09 00:48:54 +00:00
alu d5fea5e7b8 reimplement synchronization for sync methods 2025-11-09 23:59:19 +00:00
alu 79eecdf211 improve various log messages,
improve http return codes
2025-11-08 01:00:38 +00:00
alu f83a26ff6b update go mod 2025-10-10 21:55:08 +00:00
alu 612f2a159f update go mod 2025-10-02 17:46:52 +00:00
11 changed files with 672 additions and 398 deletions

No files matched your search

+3
View File
@@ -0,0 +1,3 @@
[submodule "proxmoxaas-common-lib"]
path = proxmoxaas-common-lib
url = https://git.tronnet.net/tronnet/proxmoxaas-common-lib
+2 -4
View File
@@ -1,12 +1,10 @@
.PHONY: build test clean .PHONY: build clean
build: clean build: clean
@echo "======================== Building Binary =======================" @echo "======================== Building Binary ======================="
mkdir -p dist
CGO_ENABLED=0 go build -ldflags="-s -w" -v -o dist/ . CGO_ENABLED=0 go build -ldflags="-s -w" -v -o dist/ .
test: clean
go run .
clean: clean:
@echo "======================== Cleaning Project ======================" @echo "======================== Cleaning Project ======================"
go clean go clean
+16
View File
@@ -0,0 +1,16 @@
# ProxmoxAAS Fabric - In-Memory Cache for ProxmoxAAS Cluster State
ProxmoxAAS Fabric provides functionality for the API by caching an in-memory representation of the proxmox cluster's state which speeds up computation of resource usage statistics.
## Installation
### Prerequisites
- [ProxmoxAAS-API](https://git.tronnet.net/tronnet/ProxmoxAAS-API)
- Proxmox VE Cluster (v7.0+)
### Installation - API
1. Download `proxmoxaas-fabric` binary and `template.config.json` file from releases
2. Rename `template.config.json` to `config.json` and modify the following values:
1. `listenPort`: port for proxmoxaas-fabric to bind and listen on
2. `url`: url to the Proxmox API, ie `https://pve.domain.net/api2/json`
3. `token`: the `proxmoxaas-api` API token user(name), authentication realm (pam), token id, and token secrey key (uuid)
4. `rebuildInterval` (***optional***): the interval on which to completely rebuild the cache
+99 -106
View File
@@ -6,10 +6,10 @@ import (
"fmt" "fmt"
"log" "log"
"net/http" "net/http"
paas "proxmoxaas-fabric/proxmoxaas-common-lib"
"strconv" "strconv"
"time" "time"
"github.com/gin-gonic/gin"
"github.com/luthermonson/go-proxmox" "github.com/luthermonson/go-proxmox"
) )
@@ -19,29 +19,29 @@ var client ProxmoxClient
func Run() { func Run() {
gob.Register(proxmox.Client{}) gob.Register(proxmox.Client{})
gin.SetMode(gin.ReleaseMode)
configPath := flag.String("config", "config.json", "path to config.json file") configPath := flag.String("config", "config.json", "path to config.json file")
flag.Parse() flag.Parse()
config := GetConfig(*configPath) 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) 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) client = NewClient(config.PVE.URL, token, config.PVE.Token.Secret)
router := gin.Default() router := http.NewServeMux()
cluster := Cluster{} cluster := Cluster{}
cluster.Init(client) cluster.Init(client)
start := time.Now() log.Printf("[INFO] starting cluster sync\n")
log.Printf("Starting cluster sync\n") err := cluster.Sync()
cluster.Sync() if err != nil {
log.Printf("Synced cluster in %fs\n", time.Since(start).Seconds()) log.Printf("[ERR ] error encountered while syncing cluster: %s", err)
}
// set repeating update for full rebuilds // set repeating update for full rebuilds
ticker := time.NewTicker(time.Duration(config.ReloadInterval) * time.Second) 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) channel := make(chan bool)
go func() { go func() {
for { for {
@@ -49,135 +49,128 @@ func Run() {
case <-channel: case <-channel:
return return
case <-ticker.C: case <-ticker.C:
start := time.Now() log.Printf("[INFO] starting cluster sync\n")
log.Printf("Starting cluster sync\n") err := cluster.Sync()
cluster.Sync() if err != nil {
log.Printf("Synced cluster in %fs\n", time.Since(start).Seconds()) log.Printf("[ERR ] error encountered while syncing cluster: %s", err)
}
} }
} }
}() }()
router.GET("/version", func(c *gin.Context) { router.HandleFunc("GET /version", func(w http.ResponseWriter, r *http.Request) {
PVEVersion, err := client.Version() PVEVersion, err := client.Version()
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) paas.HTTPWriteJson(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
} else { } else {
c.JSON(http.StatusOK, gin.H{"api-version": APIVersion, "pve-version": PVEVersion}) paas.HTTPWriteJson(w, http.StatusOK, map[string]any{"api-version": APIVersion, "pve-version": PVEVersion})
} }
}) })
router.GET("/nodes/:node", func(c *gin.Context) { router.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) {
nodeid := c.Param("node") callback := Callback{}
v, err := GetCluster(&callback, &cluster)
node, err := cluster.GetNode(nodeid)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) paas.HTTPWriteJson(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
} else {
paas.HTTPWriteJson(w, http.StatusOK, map[string]any{"cluster": v})
}
callback.Invoke()
})
router.HandleFunc("GET /nodes/{node}", func(w http.ResponseWriter, r *http.Request) {
nodeid := r.PathValue("node")
callback := Callback{}
node, err := GetNode(&callback, &cluster, nodeid)
if err != nil {
paas.HTTPWriteJson(w, http.StatusNotFound, map[string]any{"error": err.Error()})
} else {
paas.HTTPWriteJson(w, http.StatusOK, map[string]any{"node": node})
}
callback.Invoke()
})
router.HandleFunc("GET /nodes/{node}/instances/{vmid}", func(w http.ResponseWriter, r *http.Request) {
nodeid := r.PathValue("node")
vmid, err := strconv.ParseUint(r.PathValue("vmid"), 10, 64)
if err != nil {
paas.HTTPWriteJson(w, http.StatusBadRequest, map[string]any{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", r.PathValue("instance"))})
return
}
callback := Callback{}
instance, err := GetInstance(&callback, &cluster, nodeid, vmid)
if err != nil {
paas.HTTPWriteJson(w, http.StatusNotFound, map[string]any{"error": err.Error()})
} else {
paas.HTTPWriteJson(w, http.StatusOK, map[string]any{"instance": instance})
}
callback.Invoke()
})
router.HandleFunc("POST /sync", func(w http.ResponseWriter, r *http.Request) {
log.Printf("[INFO] starting cluster sync\n")
err := cluster.Sync()
if err != nil {
log.Printf("[ERR ] failed to sync cluster: %s", err)
paas.HTTPWriteJson(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
return return
} else { } else {
c.JSON(http.StatusOK, gin.H{"node": node})
return return
} }
}) })
router.GET("/nodes/:node/devices", func(c *gin.Context) { router.HandleFunc("POST /nodes/{node}/sync", func(w http.ResponseWriter, r *http.Request) {
nodeid := c.Param("node") nodeName := r.PathValue("node")
node, err := cluster.GetNode(nodeid) log.Printf("[INFO] starting %s sync\n", nodeName)
callback := Callback{}
node, err := GetNode(&callback, &cluster, nodeName)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error())
return paas.HTTPWriteJson(w, http.StatusNotFound, map[string]any{"error": err.Error()})
} else { callback.Invoke()
c.JSON(http.StatusOK, gin.H{"devices": node.Devices})
return return
} }
callback.Invoke()
err = node.Sync()
if err != nil {
log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error())
paas.HTTPWriteJson(w, http.StatusNotFound, map[string]any{"error": err.Error()})
}
}) })
router.GET("/nodes/:node/instances/:vmid", func(c *gin.Context) { router.HandleFunc("POST /nodes/{node}/instances/{vmid}/sync", func(w http.ResponseWriter, r *http.Request) {
nodeid := c.Param("node") nodeName := r.PathValue("node")
vmid, err := strconv.ParseUint(c.Param("vmid"), 10, 64) vmid, err := strconv.ParseUint(r.PathValue("vmid"), 10, 64)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", c.Param("instance"))}) log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
paas.HTTPWriteJson(w, http.StatusInternalServerError, map[string]any{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", r.PathValue("instance"))})
return return
} }
node, err := cluster.GetNode(nodeid) log.Printf("[INFO] starting %s.%d sync\n", nodeName, vmid)
callback := Callback{}
instance, err := GetInstance(&callback, &cluster, nodeName, vmid)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
paas.HTTPWriteJson(w, http.StatusNotFound, map[string]any{"error": err.Error()})
callback.Invoke()
return return
} else { }
instance, err := node.GetInstance(uint(vmid)) callback.Invoke()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) err = instance.Sync()
return if err != nil {
} else { log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
c.JSON(http.StatusOK, gin.H{"instance": instance}) paas.HTTPWriteJson(w, http.StatusNotFound, map[string]any{"error": err.Error()})
return
}
} }
}) })
router.POST("/sync", func(c *gin.Context) { server := paas.LoggingMiddleware(router)
//go func() {
start := time.Now()
log.Printf("Starting cluster sync\n")
cluster.Sync()
log.Printf("Synced cluster in %fs\n", time.Since(start).Seconds())
//}()
})
router.POST("/nodes/:node/sync", func(c *gin.Context) { log.Printf("[INFO] starting API listening on 0.0.0.0:%d", config.ListenPort)
nodeid := c.Param("node") log.Fatalf("[ERR ] starting gin router: %s", http.ListenAndServe("0.0.0.0:"+strconv.Itoa(config.ListenPort), server))
//go func() {
start := time.Now()
log.Printf("Starting %s sync\n", nodeid)
err := cluster.RebuildHost(nodeid)
if err != nil {
log.Printf("Failed to sync %s: %s", nodeid, err.Error())
return
} else {
log.Printf("Synced %s in %fs\n", nodeid, time.Since(start).Seconds())
return
}
//}()
})
router.POST("/nodes/:node/instances/:vmid/sync", 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"))})
return
}
//go func() {
start := time.Now()
log.Printf("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())
return
}
instance, err := node.GetInstance(uint(vmid))
if err != nil {
log.Printf("Failed to sync %s.%d: %s", nodeid, vmid, 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())
return
} else {
log.Printf("Synced %s.%d in %fs\n", nodeid, vmid, time.Since(start).Seconds())
return
}
//}()
})
router.Run("0.0.0.0:" + strconv.Itoa(config.ListenPort))
} }
+402 -122
View File
@@ -1,110 +1,308 @@
package app package app
import ( import (
"context"
"fmt" "fmt"
"log" "log"
paas "proxmoxaas-fabric/proxmoxaas-common-lib"
"strings" "strings"
"time"
"golang.org/x/sync/errgroup"
) )
func (cluster *Cluster) Init(pve ProxmoxClient) { func (cluster *Cluster) Init(pve ProxmoxClient) {
cluster.pve = pve cluster.pve = pve
} }
func GetCluster(callback *Callback, cluster *Cluster) (*Cluster, error) {
// obtain read lock on whole cluster tree
cluster.lock.RLock()
callback.Add(cluster.lock.RUnlock)
if !cluster.OK {
return nil, fmt.Errorf("cluster state is invalid")
}
for _, node := range cluster.Nodes {
node.lock.RLock()
callback.Add(node.lock.RUnlock)
for _, instance := range node.Instances {
instance.lock.RLock()
callback.Add(instance.lock.RUnlock)
}
}
return cluster, nil
}
func (cluster *Cluster) Sync() error { func (cluster *Cluster) Sync() error {
// aquire lock on cluster, release on return // obtain write lock on whole tree
cluster.lock.Lock() cluster.lock.Lock()
defer cluster.lock.Unlock() defer cluster.lock.Unlock()
for _, n := range cluster.Nodes {
n.lock.Lock()
defer n.lock.Unlock()
for _, i := range n.Instances {
i.lock.Lock()
defer i.lock.Unlock()
}
}
err := cluster.Build()
if err != nil {
cluster.OK = false
return err
}
err = cluster.ResolvePoolMembership()
if err != nil {
cluster.OK = false
return err
}
cluster.OK = true
return nil
}
func (cluster *Cluster) Build() error {
start := time.Now()
cluster.Nodes = make(map[string]*Node) cluster.Nodes = make(map[string]*Node)
wg, _ := errgroup.WithContext(context.Background())
// get all nodes // get all nodes
nodes, err := cluster.pve.Nodes() nodes, err := cluster.pve.Nodes()
if err != nil { if err != nil {
return err return err
} }
// for each node: // for each node:
for _, hostName := range nodes { for _, nodeName := range nodes {
// rebuild node wg.Go(func() error {
err := cluster.RebuildHost(hostName) // rebuild node
if err != nil { // if an error was encountered, continue and log the error node := Node{}
log.Print(err.Error())
continue node.Name = nodeName
} node.cluster = cluster
err := node.Build()
if err != nil { // if an error was encountered, continue and log the error
log.Printf("[ERR ] error encountered while syncing node %s: %s", nodeName, err)
}
cluster.NodesLock.Lock()
cluster.Nodes[nodeName] = &node
cluster.NodesLock.Unlock()
return err
})
} }
err = wg.Wait()
if err != nil {
cluster.lock.Unlock()
return err
}
log.Printf("[INFO] built cluster in %d ms", time.Since(start).Milliseconds())
return nil return nil
} }
// get a node in the cluster func (cluster *Cluster) ResolvePoolMembership() error {
func (cluster *Cluster) GetNode(hostName string) (*Node, error) { start := time.Now()
host_ch := make(chan *Node)
err_ch := make(chan error)
go func() { //clear existing pool memberships
// aquire cluster lock for _, node := range cluster.Nodes {
cluster.lock.Lock() for _, instance := range node.Instances {
defer cluster.lock.Unlock() instance.Pool = ""
// get host
host, ok := cluster.Nodes[hostName]
if !ok {
host_ch <- nil
err_ch <- fmt.Errorf("%s not in cluster", hostName)
} else {
// aquire host lock to wait in case of a concurrent write
host.lock.Lock()
defer host.lock.Unlock()
host_ch <- host
err_ch <- nil
} }
}()
host := <-host_ch
err := <-err_ch
return host, err
}
func (cluster *Cluster) RebuildHost(hostName string) error {
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())
} }
// aquire lock on host, release on return //resolve pool membership
host.lock.Lock() pools, err := cluster.pve.client.Pools(context.Background())
defer host.lock.Unlock() if err != nil {
return err
}
for _, pool := range pools {
pool, err = cluster.pve.client.Pool(context.Background(), pool.PoolID)
if err != nil {
return err
}
for _, member := range pool.Members {
if member.Type == "lxc" || member.Type == "qemu" {
node, ok := cluster.Nodes[member.Node]
if !ok {
return fmt.Errorf("Instance %d has no node", member.VMID)
}
instance, ok := node.Instances[InstanceID(member.VMID)]
if !ok {
return fmt.Errorf("Instance %d claimed to be in node %s but was not", member.VMID, node.Name)
}
if instance.Pool != "" { // enforces that an instance cannot be in two different pools
return fmt.Errorf("Instance %d is in pools %s and %s which is not supported", member.VMID, instance.Pool, pool.PoolID)
}
instance.Pool = pool.PoolID
log.Printf("[INFO] resolved pool membership for vmid=%d pool=%s", member.VMID, pool.PoolID)
}
}
}
cluster.Nodes[hostName] = host log.Printf("[INFO] resovled cluster instance pool memberships in %d ms", time.Since(start).Milliseconds())
return nil
}
func GetNode(callback *Callback, cluster *Cluster, nodeName string) (*Node, error) {
cluster.lock.RLock()
callback.Add(cluster.lock.RUnlock)
if !cluster.OK {
return nil, fmt.Errorf("cluster state is invalid")
}
// get node
node, ok := cluster.Nodes[nodeName]
if !ok {
return nil, fmt.Errorf("%s not in cluster", nodeName)
}
node.lock.RLock()
callback.Add(node.lock.RUnlock)
for _, instance := range node.Instances {
instance.lock.RLock()
callback.Add(instance.lock.RUnlock)
}
return node, nil
}
func (node *Node) Sync() error {
cluster := node.cluster
// obtain write lock on this node subtree, and obtain write lock all instances for ResolvePoolMembership, obtain read lock on everything else
cluster.lock.Lock() // todo relax
defer cluster.lock.Unlock()
for _, n := range cluster.Nodes {
if n != node { // if other node, read lock
n.lock.RLock()
defer n.lock.RUnlock()
} else { // if this node, write lock
n.lock.Lock()
defer n.lock.Unlock()
}
for _, i := range n.Instances { // need write lock on all instances
i.lock.Lock()
defer i.lock.Unlock()
}
}
err := node.Build()
if err != nil {
cluster.OK = false
return err
}
err = cluster.ResolvePoolMembership()
if err != nil {
cluster.OK = false
return err
}
cluster.OK = true
return nil
}
func (node *Node) Build() error {
start := time.Now()
nodeName := node.Name
cluster := node.cluster
err := cluster.pve.Node(node, nodeName)
if err != nil && cluster.Nodes[nodeName] == nil { // node is unreachable and did not exist previously
// return an error because we requested to sync a node that was not already in the cluster
return fmt.Errorf("error retrieving %s: %s", nodeName, err.Error())
}
wg, _ := errgroup.WithContext(context.Background())
if err != nil && cluster.Nodes[nodeName] != nil { // node is unreachable and did exist previously
// assume the node is down or gone and delete from cluster
delete(cluster.Nodes, nodeName)
return nil
}
// get node's VMs // get node's VMs
vms, err := host.VirtualMachines() vms, err := node.VirtualMachines()
if err != nil { if err != nil {
return err return err
} }
for _, vmid := range vms { for _, vmid := range vms {
err := host.RebuildInstance(VM, vmid) wg.Go(func() error {
if err != nil { // if an error was encountered, continue and log the error
log.Print(err.Error()) instanceID := InstanceID(paas.SafeUint64(vmid))
continue instance := Instance{}
} instance.lock.Lock()
instance.VMID = instanceID
instance.Type = VM
instance.node = node
err := instance.Build()
if err != nil { // if an error was encountered, continue and log the error
log.Printf("[ERR ] error encountered while syncing vm %s.%d: %s", nodeName, vmid, err)
}
node.Instances[instanceID] = &instance
instance.lock.Unlock()
return nil
})
} }
// get node's CTs // get node's CTs
cts, err := host.Containers() cts, err := node.Containers()
if err != nil { if err != nil {
return err return err
} }
for _, vmid := range cts { for _, vmid := range cts {
err := host.RebuildInstance(CT, vmid) wg.Go(func() error {
if err != nil { instanceID := InstanceID(paas.SafeUint64(vmid))
return err instance := Instance{}
} instance.lock.Lock()
instance.VMID = instanceID
instance.Type = CT
instance.node = node
err := instance.Build()
if err != nil { // if an error was encountered, continue and log the error
log.Printf("[ERR ] error encountered while syncing ct %s.%d: %s", nodeName, vmid, err)
}
node.InstancesLock.Lock()
node.Instances[instanceID] = &instance
node.InstancesLock.Unlock()
instance.lock.Unlock()
return nil
})
}
err = wg.Wait()
if err != nil {
return err
} }
// check node device reserved by iterating over each function, we will assume that a single reserved function means the device is also reserved // check node device reserved by iterating over each function, we will assume that a single reserved function means the device is also reserved
for _, device := range host.Devices { for _, device := range node.Devices {
reserved := false reserved := false
for _, function := range device.Functions { for _, function := range device.Functions {
reserved = reserved || function.Reserved reserved = reserved || function.Reserved
@@ -112,96 +310,172 @@ func (cluster *Cluster) RebuildHost(hostName string) error {
device.Reserved = reserved device.Reserved = reserved
} }
node.cluster = cluster
log.Printf("[INFO] built node %s in %d ms", node.Name, time.Since(start).Milliseconds())
return nil return nil
} }
func (host *Node) GetInstance(vmid uint) (*Instance, error) { func GetInstance(callback *Callback, cluster *Cluster, nodeName string, vmid uint64) (*Instance, error) {
instance_ch := make(chan *Instance) // obtain read lock on whole tree, but do not hold cluster and node read lock indefinitely
err_ch := make(chan error) cluster.lock.RLock()
callback.Add(cluster.lock.RUnlock)
go func() { if !cluster.OK {
// aquire host lock return nil, fmt.Errorf("cluster state is invalid")
host.lock.Lock() }
defer host.lock.Unlock()
// get instance
instance, ok := host.Instances[InstanceID(vmid)]
if !ok {
instance_ch <- nil
err_ch <- fmt.Errorf("vmid %d not in host %s", vmid, host.Name)
} else {
// aquire instance lock to wait in case of a concurrent write
instance.lock.Lock()
defer instance.lock.Unlock()
instance_ch <- instance // get node
err_ch <- nil node, ok := cluster.Nodes[nodeName]
} if !ok {
}() return nil, fmt.Errorf("%s not in cluster", nodeName)
}
node.lock.RLock()
callback.Add(node.lock.RUnlock)
instance := <-instance_ch // get instance
err := <-err_ch instance, ok := node.Instances[InstanceID(vmid)]
return instance, err if !ok {
return nil, fmt.Errorf("vmid %d not in node %s", vmid, node.Name)
} else {
instance.lock.RLock()
callback.Add(instance.lock.RUnlock)
return instance, nil
}
} }
func (host *Node) RebuildInstance(instancetype InstanceType, vmid uint) error { func (instance *Instance) Sync() error {
var instance *Instance cluster := instance.node.cluster
if instancetype == VM {
var err error // obtain write lock on this instance subtree, and obtain write lock all instances for ResolvePoolMembership, obtain read lock on everything else
instance, err = host.VirtualMachine(vmid) cluster.lock.Lock() // todo relax
if err != nil { defer cluster.lock.Unlock()
return fmt.Errorf("error retrieving %d: %s, possibly down?", vmid, err.Error()) for _, n := range cluster.Nodes {
} n.lock.RLock()
} else if instancetype == CT { defer n.lock.RUnlock()
var err error for _, i := range n.Instances { // need write lock on all instances
instance, err = host.Container(vmid) i.lock.Lock()
if err != nil { defer i.lock.Unlock()
return fmt.Errorf("error retrieving %d: %s, possibly down?", vmid, err.Error())
} }
}
err := instance.Build()
if err != nil {
cluster.OK = false
return err
}
err = cluster.ResolvePoolMembership()
if err != nil {
cluster.OK = false
return err
}
cluster.OK = true
return nil
}
func (instance *Instance) Build() error {
start := time.Now()
vmid := instance.VMID
instancetype := instance.Type
node := instance.node
var err error
switch instancetype {
case VM:
err = node.VirtualMachine(instance, vmid)
case CT:
err = node.Container(instance, vmid)
} }
// aquire lock on instance, release on return if err != nil && node.Instances[vmid] == nil { // instance is unreachable and did not exist previously
instance.lock.Lock() // return an error because we requested to sync an instance that was not already in the cluster
defer instance.lock.Unlock() return fmt.Errorf("error retrieving %s.%d: %s", node.Name, vmid, err.Error())
}
host.Instances[InstanceID(vmid)] = instance wg, _ := errgroup.WithContext(context.Background())
if err != nil && node.Instances[vmid] != nil { // node is unreachable and did exist previously
// assume the instance is gone and delete from cluster
log.Printf("[ERR ] error retrieving %s.%d: %s", node.Name, vmid, err)
delete(node.Instances, vmid)
return nil
}
for volid := range instance.configDisks { for volid := range instance.configDisks {
instance.RebuildVolume(host, volid) wg.Go(func() error {
err := instance.BuildVolume(node, volid)
if err != nil {
log.Printf("[ERR ] error rebuilding volume %s: %s", volid, err)
}
return err
})
} }
for netid := range instance.configNets { for netid := range instance.configNets {
instance.RebuildNet(netid) wg.Go(func() error {
err := instance.BuildNet(node, netid)
if err != nil {
log.Printf("[ERR ] error rebuilding net %s: %s", netid, err)
return err
}
return err
})
} }
for deviceid := range instance.configHostPCIs { for deviceid := range instance.configHostPCIs {
instance.RebuildDevice(host, deviceid) wg.Go(func() error {
err := instance.BuildDevice(node, deviceid)
if err != nil {
log.Printf("[ERR ] error rebuilding pci %s: %s", deviceid, err)
}
return err
})
} }
if instance.Type == VM { err = wg.Wait()
instance.RebuildBoot()
}
return nil
}
func (instance *Instance) RebuildVolume(host *Node, volid string) error {
volumeDataString := instance.configDisks[volid]
volume, err := GetVolumeInfo(host, volumeDataString)
if err != nil { if err != nil {
return err return err
} }
voltype := AnyPrefixes(volid, VolumeTypes) if instance.Type == VM {
volume.Type = voltype err := instance.BuildBoot(node)
volume.Volume_ID = VolumeID(volid) if err != nil {
instance.Volumes[VolumeID(volid)] = volume log.Printf("[ERR ] error rebuilding boot: %s", err)
return err
}
}
instance.node = node
log.Printf("[INFO] built instance %s.%d in %d ms", instance.node.Name, instance.VMID, time.Since(start).Milliseconds())
return nil return nil
} }
func (instance *Instance) RebuildNet(netid string) error { func (instance *Instance) BuildVolume(node *Node, volid string) error {
volumeDataString := instance.configDisks[volid]
volume, err := GetVolumeInfo(node, volumeDataString)
if err != nil {
return err
}
voltype := AnyPrefixes(volid, paas.VolumeTypes)
volume.Type = voltype
volume.Volume_ID = VolumeID(volid)
instance.VolumesLock.Lock()
instance.Volumes[VolumeID(volid)] = volume
instance.VolumesLock.Unlock()
return nil
}
func (instance *Instance) BuildNet(node *Node, netid string) error {
net := instance.configNets[netid] net := instance.configNets[netid]
netinfo, err := GetNetInfo(net) netinfo, err := GetNetInfo(net)
@@ -210,22 +484,25 @@ func (instance *Instance) RebuildNet(netid string) error {
return nil return nil
} }
instance.NetsLock.Lock()
instance.Nets[NetID(netid)] = netinfo instance.Nets[NetID(netid)] = netinfo
instance.NetsLock.Unlock()
return nil return nil
} }
func (instance *Instance) RebuildDevice(host *Node, deviceid string) error { func (instance *Instance) BuildDevice(node *Node, deviceid string) error {
instanceDevice, ok := instance.configHostPCIs[deviceid] instanceDevice, ok := instance.configHostPCIs[deviceid]
if !ok { // if device does not exist if !ok { // if device does not exist
return fmt.Errorf("%s not found in devices", deviceid) log.Printf("[WARN] %s not found in devices on node %s", deviceid, node.Name)
return nil
} }
hostDeviceBusID := DeviceID(strings.Split(instanceDevice, ",")[0]) hostDeviceBusID := DeviceID(strings.Split(instanceDevice, ",")[0])
instanceDeviceBusID := DeviceID(deviceid) instanceDeviceBusID := DeviceID(deviceid)
if DeviceBusIDIsSuperDevice(hostDeviceBusID) { if DeviceBusIDIsSuperDevice(hostDeviceBusID) {
instance.Devices[DeviceID(instanceDeviceBusID)] = host.Devices[DeviceBus(hostDeviceBusID)] instance.Devices[DeviceID(instanceDeviceBusID)] = node.Devices[DeviceBus(hostDeviceBusID)]
for _, function := range instance.Devices[DeviceID(instanceDeviceBusID)].Functions { for _, function := range instance.Devices[DeviceID(instanceDeviceBusID)].Functions {
function.Reserved = true function.Reserved = true
} }
@@ -233,13 +510,14 @@ func (instance *Instance) RebuildDevice(host *Node, deviceid string) error {
// sub function assignment not supported yet // sub function assignment not supported yet
} }
instance.DevicesLock.Lock()
instance.Devices[DeviceID(instanceDeviceBusID)].Device_ID = DeviceID(deviceid) instance.Devices[DeviceID(instanceDeviceBusID)].Device_ID = DeviceID(deviceid)
instance.Devices[DeviceID(instanceDeviceBusID)].Value = instanceDevice instance.DevicesLock.Unlock()
return nil return nil
} }
func (instance *Instance) RebuildBoot() { func (instance *Instance) BuildBoot(node *Node) error {
instance.Boot = BootOrder{} instance.Boot = BootOrder{}
eligibleBoot := map[string]bool{} eligibleBoot := map[string]bool{}
@@ -265,7 +543,7 @@ func (instance *Instance) RebuildBoot() {
instance.Boot.Enabled = append(instance.Boot.Enabled, val) instance.Boot.Enabled = append(instance.Boot.Enabled, val)
delete(eligibleBoot, bootTarget) delete(eligibleBoot, bootTarget)
} else { // item is not eligible for boot but is included in the boot order } 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) delete(eligibleBoot, bootTarget)
} }
} }
@@ -277,7 +555,9 @@ func (instance *Instance) RebuildBoot() {
} else if val, ok := instance.Nets[NetID(bootTarget)]; ok && isEligible { // if the item is eligible and is in nets } 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) 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 } 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)
} }
} }
return nil
} }
+71 -48
View File
@@ -4,11 +4,14 @@ import (
"context" "context"
"crypto/tls" "crypto/tls"
"fmt" "fmt"
"maps"
"net/http" "net/http"
paas "proxmoxaas-fabric/proxmoxaas-common-lib"
"strconv" "strconv"
"strings" "strings"
"github.com/luthermonson/go-proxmox" "github.com/luthermonson/go-proxmox"
"golang.org/x/sync/errgroup"
) )
type ProxmoxClient struct { type ProxmoxClient struct {
@@ -32,9 +35,7 @@ type PVEProctype struct {
func NewClient(url string, token string, secret string) ProxmoxClient { func NewClient(url string, token string, secret string) ProxmoxClient {
HTTPClient := http.Client{ HTTPClient := http.Client{
Transport: &http.Transport{ Transport: &http.Transport{
TLSClientConfig: &tls.Config{ TLSClientConfig: &tls.Config{},
InsecureSkipVerify: true,
},
}, },
} }
@@ -71,22 +72,25 @@ func (pve ProxmoxClient) Nodes() ([]string, error) {
} }
// Gets a Node's resources but does not recursively expand instances // Gets a Node's resources but does not recursively expand instances
func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { func (pve ProxmoxClient) Node(host *Node, nodeName string) error {
host := Node{} //host := Node{}
host.Devices = make(map[DeviceBus]*Device) host.Devices = make(map[DeviceBus]*Device)
host.Instances = make(map[InstanceID]*Instance) host.Instances = make(map[InstanceID]*Instance)
host.storage = make(map[string][]*proxmox.StorageContent)
// get pve node
node, err := pve.client.Node(context.Background(), nodeName) node, err := pve.client.Node(context.Background(), nodeName)
if err != nil { if err != nil {
return &host, err return err
} }
// get pve node devices
devices := []PVEDevice{} devices := []PVEDevice{}
err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/hardware/pci", nodeName), &devices) err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/hardware/pci", nodeName), &devices)
if err != nil { if err != nil {
return &host, err return err
} }
// populate host devices from pve node devices
for _, device := range devices { for _, device := range devices {
x := strings.Split(device.ID, ".") x := strings.Split(device.ID, ".")
if len(x) != 2 { // this should always be true, but skip if not if len(x) != 2 { // this should always be true, but skip if not
@@ -110,22 +114,51 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) {
} }
} }
// get pve node proctypes
proctypes := []PVEProctype{} proctypes := []PVEProctype{}
err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/capabilities/qemu/cpu", nodeName), &proctypes) err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/capabilities/qemu/cpu", nodeName), &proctypes)
if err != nil { if err != nil {
return &host, err return err
} }
// populate host proctypes from pve node proctypes
for _, proctype := range proctypes { for _, proctype := range proctypes {
host.Proctypes = append(host.Proctypes, proctype.Name) host.Proctypes = append(host.Proctypes, proctype.Name)
} }
// get all pve node storages
wg, _ := errgroup.WithContext(context.Background())
storages, err := node.Storages(context.Background())
if err != nil {
return err
}
// populate host storage content from pve node storages
for _, storage := range storages {
wg.Go(func() error {
if storage.Enabled == 1 && storage.Active == 1 {
content, err := storage.GetContent(context.Background())
if err != nil {
return err
}
host.storageLock.Lock()
host.storage[storage.Name] = content
host.storageLock.Unlock()
}
return nil
})
}
err = wg.Wait()
if err != nil {
return err
}
// set host basic values
host.Name = node.Name host.Name = node.Name
host.Cores = uint64(node.CPUInfo.CPUs) host.Cores = paas.SafeUint64(node.CPUInfo.CPUs)
host.Memory = uint64(node.Memory.Total) host.Memory = node.Memory.Total
host.Swap = uint64(node.Swap.Total) host.Swap = node.Swap.Total
host.pvenode = node host.pvenode = node
return &host, err return err
} }
// Get all VM IDs on specified host // Get all VM IDs on specified host
@@ -142,16 +175,15 @@ func (host *Node) VirtualMachines() ([]uint, error) {
} }
// Get a VM's CPU, Memory but does not recursively link Devices, Disks, Drives, Nets // Get a VM's CPU, Memory but does not recursively link Devices, Disks, Drives, Nets
func (host *Node) VirtualMachine(VMID uint) (*Instance, error) { func (host *Node) VirtualMachine(instance *Instance, VMID InstanceID) error {
instance := Instance{}
vm, err := host.pvenode.VirtualMachine(context.Background(), int(VMID)) vm, err := host.pvenode.VirtualMachine(context.Background(), int(VMID))
if err != nil { if err != nil {
return &instance, err return err
} }
config := vm.VirtualMachineConfig config := vm.VirtualMachineConfig
instance.configHostPCIs = config.MergeHostPCIs() instance.configHostPCIs = config.HostPCIs
instance.configNets = config.MergeNets() instance.configNets = config.Nets
instance.configDisks = MergeVMDisksAndUnused(config) instance.configDisks = MergeVMDisksAndUnused(config)
instance.configBoot = config.Boot instance.configBoot = config.Boot
@@ -160,20 +192,18 @@ func (host *Node) VirtualMachine(VMID uint) (*Instance, error) {
instance.Name = vm.Name instance.Name = vm.Name
instance.Proctype = vm.VirtualMachineConfig.CPU instance.Proctype = vm.VirtualMachineConfig.CPU
instance.Cores = uint64(vm.VirtualMachineConfig.Cores) instance.Cores = paas.SafeUint64(*vm.VirtualMachineConfig.Cores)
instance.Memory = uint64(vm.VirtualMachineConfig.Memory) * MiB instance.Memory = paas.SafeUint64(int(vm.VirtualMachineConfig.Memory)) * paas.MiB
instance.Volumes = make(map[VolumeID]*Volume) instance.Volumes = make(map[VolumeID]*Volume)
instance.Nets = make(map[NetID]*Net) instance.Nets = make(map[NetID]*Net)
instance.Devices = make(map[DeviceID]*Device) instance.Devices = make(map[DeviceID]*Device)
return &instance, nil return nil
} }
func MergeVMDisksAndUnused(vmc *proxmox.VirtualMachineConfig) map[string]string { func MergeVMDisksAndUnused(vmc *proxmox.VirtualMachineConfig) map[string]string {
mergedDisks := vmc.MergeDisks() mergedDisks := vmc.MergeDisks()
for k, v := range vmc.MergeUnuseds() { maps.Copy(mergedDisks, vmc.Unuseds)
mergedDisks[k] = v
}
return mergedDisks return mergedDisks
} }
@@ -191,39 +221,34 @@ func (host *Node) Containers() ([]uint, error) {
} }
// Get a CT's CPU, Memory, Swap but does not recursively link Devices, Disks, Drives, Nets // Get a CT's CPU, Memory, Swap but does not recursively link Devices, Disks, Drives, Nets
func (host *Node) Container(VMID uint) (*Instance, error) { func (host *Node) Container(instance *Instance, VMID InstanceID) error {
instance := Instance{}
ct, err := host.pvenode.Container(context.Background(), int(VMID)) ct, err := host.pvenode.Container(context.Background(), int(VMID))
if err != nil { if err != nil {
return &instance, err return err
} }
config := ct.ContainerConfig config := ct.ContainerConfig
instance.configHostPCIs = make(map[string]string) instance.configHostPCIs = make(map[string]string)
instance.configNets = config.MergeNets() instance.configNets = config.Nets
instance.configDisks = MergeCTDisksAndUnused(config) instance.configDisks = MergeCTDisksAndUnused(config)
instance.pveconfig = config instance.pveconfig = config
instance.Type = CT instance.Type = CT
instance.Name = ct.Name instance.Name = ct.Name
instance.Cores = uint64(ct.ContainerConfig.Cores) instance.Cores = paas.SafeUint64(ct.ContainerConfig.Cores)
instance.Memory = uint64(ct.ContainerConfig.Memory) * MiB instance.Memory = paas.SafeUint64(*ct.ContainerConfig.Memory) * paas.MiB
instance.Swap = uint64(ct.ContainerConfig.Swap) * MiB instance.Swap = paas.SafeUint64(*ct.ContainerConfig.Swap) * paas.MiB
instance.Volumes = make(map[VolumeID]*Volume) instance.Volumes = make(map[VolumeID]*Volume)
instance.Nets = make(map[NetID]*Net) instance.Nets = make(map[NetID]*Net)
return &instance, nil return nil
} }
func MergeCTDisksAndUnused(cc *proxmox.ContainerConfig) map[string]string { func MergeCTDisksAndUnused(cc *proxmox.ContainerConfig) map[string]string {
mergedDisks := make(map[string]string) mergedDisks := make(map[string]string)
for k, v := range cc.MergeUnuseds() { maps.Copy(mergedDisks, cc.Unuseds)
mergedDisks[k] = v maps.Copy(mergedDisks, cc.Mps)
}
for k, v := range cc.MergeMps() {
mergedDisks[k] = v
}
mergedDisks["rootfs"] = cc.RootFS mergedDisks["rootfs"] = cc.RootFS
return mergedDisks return mergedDisks
} }
@@ -236,21 +261,19 @@ func GetVolumeInfo(host *Node, volume string) (*Volume, error) {
volumeFile := volumeObj[""] volumeFile := volumeObj[""]
volumeStorage := strings.Split(volumeFile, ":")[0] volumeStorage := strings.Split(volumeFile, ":")[0]
storage, err := host.pvenode.Storage(context.Background(), volumeStorage) host.storageLock.Lock()
if err != nil { storage, ok := host.storage[volumeStorage]
return &volumeData, nil host.storageLock.Unlock()
if !ok {
return &volumeData, fmt.Errorf("volume %s claims to be in storage %s but storage was not found", volume, volumeStorage)
} }
content, err := storage.GetContent(context.Background()) for _, c := range storage {
if err != nil {
return &volumeData, nil
}
for _, c := range content {
if c.Volid == volumeFile { if c.Volid == volumeFile {
volumeData.Storage = volumeStorage volumeData.Storage = volumeStorage
volumeData.Format = c.Format volumeData.Format = c.Format
volumeData.Size = uint64(c.Size) volumeData.Size = c.Size
volumeData.File = volumeFile volumeData.File = volumeFile
volumeData.MP = volumeObj["mp"] volumeData.MP = volumeObj["mp"]
} }
+52 -76
View File
@@ -1,105 +1,81 @@
package app package app
import ( import (
paas "proxmoxaas-fabric/proxmoxaas-common-lib"
"sync" "sync"
"github.com/luthermonson/go-proxmox" "github.com/luthermonson/go-proxmox"
) )
// uses and aliases common resource types from proxmoxaas-common-lib
// except for Cluster, Node, and Instance which require mutex locks and additional fields
// add mutex and pve client
// override Nodes map to new custom Node type
type Cluster struct { type Cluster struct {
lock sync.Mutex paas.Cluster
pve ProxmoxClient lock sync.RWMutex
Nodes map[string]*Node pve ProxmoxClient
NodesLock sync.Mutex // lock for Nodes map
Nodes map[string]*Node `json:"nodes"`
OK bool
} }
// add mutex and pve api Node object
// override Instances map to new custom Instance type
type Node struct { type Node struct {
lock sync.Mutex paas.Node
Name string `json:"name"` lock sync.RWMutex
Cores uint64 `json:"cores"` InstancesLock sync.Mutex // lock for Instances map
Memory uint64 `json:"memory"` Instances map[InstanceID]*Instance `json:"instances"`
Swap uint64 `json:"swap"` pvenode *proxmox.Node
Devices map[DeviceBus]*Device `json:"devices"` storageLock sync.Mutex
Instances map[InstanceID]*Instance `json:"instances"` storage map[string][]*proxmox.StorageContent
Proctypes []string `json:"cpus"` cluster *Cluster
pvenode *proxmox.Node
} }
type InstanceID uint64 type InstanceID = paas.InstanceID
type InstanceType string type InstanceType = paas.InstanceType
const ( const VM InstanceType = paas.VM
VM InstanceType = "VM" const CT InstanceType = paas.CT
CT InstanceType = "CT"
)
// add mutex and various config objects
type Instance struct { type Instance struct {
lock sync.Mutex paas.Instance
Type InstanceType `json:"type"` lock sync.RWMutex
Name string `json:"name"` VolumesLock sync.Mutex // lock for Volumes map
Proctype string `json:"cpu"` NetsLock sync.Mutex // lock for Nets map
Cores uint64 `json:"cores"` DevicesLock sync.Mutex // lock for Devices map
Memory uint64 `json:"memory"`
Swap uint64 `json:"swap"`
Volumes map[VolumeID]*Volume `json:"volumes"`
Nets map[NetID]*Net `json:"nets"`
Devices map[DeviceID]*Device `json:"devices"`
Boot BootOrder `json:"boot"`
pveconfig any pveconfig any
configDisks map[string]string configDisks map[string]string
configNets map[string]string configNets map[string]string
configHostPCIs map[string]string configHostPCIs map[string]string
configBoot string configBoot string
node *Node
} }
var VolumeTypes = []string{ type VolumeID = paas.VolumeID
"sata", type Volume = paas.Volume
"scsi", type NetID = paas.NetID
"ide", type Net = paas.Net
"rootfs", type DeviceID = paas.DeviceID
"mp", type DeviceBus = paas.DeviceBus
"unused", type Device = paas.Device
type FunctionID = paas.FunctionID
type Function = paas.Function
type BootOrder = paas.BootOrder
type Callback struct {
f []func()
} }
type VolumeID string func (c *Callback) Add(nf func()) {
type Volume struct { c.f = append(c.f, nf)
Volume_ID VolumeID `json:"volume_id"`
Type string `json:"type"`
Storage string `json:"storage"`
Format string `json:"format"`
Size uint64 `json:"size"`
File string `json:"file"`
MP string `json:"mp"`
} }
type NetID string func (c *Callback) Invoke() {
type Net struct { for _, f := range c.f {
Net_ID NetID `json:"net_id"` f()
Value string `json:"value"` }
Rate uint64 `json:"rate"`
VLAN uint64 `json:"vlan"`
}
type DeviceID string
type DeviceBus string
type Device struct {
Device_ID DeviceID `json:"device_id"`
Device_Bus DeviceBus `json:"device_bus"`
Device_Name string `json:"device_name"`
Vendor_Name string `json:"vendor_name"`
Functions map[FunctionID]*Function `json:"functions"`
Reserved bool `json:"reserved"`
Value string
}
type FunctionID string
type Function struct {
Function_ID FunctionID `json:"function_id"`
Function_Name string `json:"subsystem_device_name"`
Vendor_Name string `json:"subsystem_vendor_name"`
Reserved bool `json:"reserved"`
}
type BootOrder struct {
Enabled []any `json:"enabled"`
Disabled []any `json:"disabled"`
} }
+11 -5
View File
@@ -7,15 +7,13 @@ import (
"strings" "strings"
) )
const MiB = 1024 * 1024
type Config struct { type Config struct {
ListenPort int `json:"listenPort"` ListenPort int `json:"listenPort"`
PVE struct { PVE struct {
URL string `json:"url"` URL string `json:"url"`
Token struct { Token struct {
USER string `json:"user"` User string `json:"user"`
REALM string `json:"realm"` Realm string `json:"realm"`
ID string `json:"id"` ID string `json:"id"`
Secret string `json:"uuid"` Secret string `json:"uuid"`
} }
@@ -24,15 +22,23 @@ type Config struct {
} }
func GetConfig(configPath string) Config { func GetConfig(configPath string) Config {
content, err := os.ReadFile(configPath) root, err := os.OpenRoot(".")
if err != nil {
log.Fatal("Error when opening root dir: ", err)
}
defer root.Close()
content, err := root.ReadFile(configPath)
if err != nil { if err != nil {
log.Fatal("Error when opening config file: ", err) log.Fatal("Error when opening config file: ", err)
} }
var config Config var config Config
err = json.Unmarshal(content, &config) err = json.Unmarshal(content, &config)
if err != nil { if err != nil {
log.Fatal("Error during parsing config file: ", err) log.Fatal("Error during parsing config file: ", err)
} }
return config return config
} }
+2 -2
View File
@@ -1,7 +1,7 @@
{ {
"listenPort": 80, "listenPort": 8082,
"pve": { "pve": {
"url": "http://<proxmox host>/api2/json", "url": "https://<proxmox host>/api2/json",
"token": { "token": {
"user": "proxmoxaas-api", "user": "proxmoxaas-api",
"realm": "pam", "realm": "pam",
+13 -35
View File
@@ -1,49 +1,27 @@
module proxmoxaas-fabric module proxmoxaas-fabric
go 1.25.1 go 1.27.3
require ( require (
github.com/gin-gonic/gin v1.10.1 github.com/luthermonson/go-proxmox v0.8.2
github.com/luthermonson/go-proxmox v0.2.3 golang.org/x/sync v0.24.0
) )
require ( require (
github.com/anchore/go-lzo v0.1.0 // indirect
github.com/buger/goterm v1.0.4 // indirect github.com/buger/goterm v1.0.4 // indirect
github.com/bytedance/gopkg v0.1.3 // indirect github.com/diskfs/go-diskfs v1.9.4 // indirect
github.com/bytedance/sonic v1.14.1 // indirect
github.com/bytedance/sonic/loader v0.3.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/djherbis/times v1.6.0 // indirect
github.com/elliotwutingfeng/asciiset v0.0.0-20250812055617-fb43ac3ba420 // indirect github.com/elliotwutingfeng/asciiset v0.0.0-20260129054604-cfde2086bc57 // indirect
github.com/gabriel-vasile/mimetype v1.4.10 // 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/goccy/go-json v0.10.5 // indirect
github.com/google/uuid v1.6.0 // indirect github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect github.com/gorilla/websocket v1.5.3 // indirect
github.com/jinzhu/copier v0.4.0 // indirect github.com/jinzhu/copier v0.4.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.18.5 // indirect
github.com/klauspost/compress v1.18.0 // indirect github.com/magefile/mage v1.17.2 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/pierrec/lz4/v4 v4.1.26 // indirect
github.com/leodido/go-urn v1.4.0 // indirect
github.com/magefile/mage v1.15.0 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
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/pkg/xattr v0.4.12 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/sirupsen/logrus v1.9.4 // indirect
github.com/ugorji/go/codec v1.3.0 // indirect github.com/stretchr/testify v1.12.1 // indirect
github.com/ulikunitz/xz v0.5.12 // indirect github.com/ulikunitz/xz v0.5.15 // indirect
golang.org/x/arch v0.21.0 // indirect golang.org/x/sys v0.49.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
) )