fix possible erroneous return if sync and fetch occur simultaneously

This commit is contained in:
2026-07-07 03:37:48 +00:00
parent a7b932f600
commit 54e345a308
4 changed files with 41 additions and 42 deletions
+5 -26
View File
@@ -76,11 +76,6 @@ func Run() {
}) })
router.GET("/", func(c *gin.Context) { router.GET("/", func(c *gin.Context) {
if !cluster.OK {
c.JSON(http.StatusInternalServerError, gin.H{"error": "cluster state is invalid"})
return
}
v, err := cluster.Get() v, err := cluster.Get()
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
@@ -92,11 +87,6 @@ func Run() {
}) })
router.GET("/nodes/:node", func(c *gin.Context) { router.GET("/nodes/:node", func(c *gin.Context) {
if !cluster.OK {
c.JSON(http.StatusInternalServerError, gin.H{"error": "cluster state is invalid"})
return
}
nodeid := c.Param("node") nodeid := c.Param("node")
node, err := cluster.GetNode(nodeid) node, err := cluster.GetNode(nodeid)
if err != nil { if err != nil {
@@ -109,30 +99,19 @@ func Run() {
}) })
router.GET("/nodes/:node/instances/:vmid", func(c *gin.Context) { router.GET("/nodes/:node/instances/:vmid", func(c *gin.Context) {
if !cluster.OK {
c.JSON(http.StatusInternalServerError, gin.H{"error": "cluster state is invalid"})
return
}
nodeid := c.Param("node") nodeid := c.Param("node")
vmid, err := strconv.ParseUint(c.Param("vmid"), 10, 64) vmid, err := strconv.ParseUint(c.Param("vmid"), 10, 64)
if err != nil { if err != nil {
c.JSON(http.StatusBadRequest, 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 return
} }
node, err := cluster.GetNode(nodeid) instance, err := cluster.GetInstance(nodeid, vmid)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return return
} else { } else {
instance, err := node.GetInstance(uint(vmid)) c.JSON(http.StatusOK, gin.H{"instance": instance})
if err != nil { return
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
} else {
c.JSON(http.StatusOK, gin.H{"instance": instance})
return
}
} }
}) })
@@ -172,7 +151,7 @@ func Run() {
} }
start := time.Now() start := time.Now()
log.Printf("[INFO] starting %s.%d sync\n", nodeName, vmid) log.Printf("[INFO] starting %s.%d sync\n", nodeName, vmid)
err = SyncInstance(&cluster, nodeName, uint(vmid)) err = SyncInstance(&cluster, nodeName, vmid)
if err != nil { if err != nil {
log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error()) log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+33 -13
View File
@@ -28,15 +28,15 @@ func (cluster *Cluster) Get() (*Cluster, error) {
} }
func SyncCluster(cluster *Cluster) error { func SyncCluster(cluster *Cluster) error {
cluster.OK = false
err := cluster.BuildCluster() err := cluster.BuildCluster()
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
err = cluster.ResolvePoolMembership() err = cluster.ResolvePoolMembership()
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
@@ -135,6 +135,10 @@ func (cluster *Cluster) GetNode(nodeName string) (*Node, error) {
cluster.lock.Lock() cluster.lock.Lock()
defer cluster.lock.Unlock() defer cluster.lock.Unlock()
if !cluster.OK {
return nil, fmt.Errorf("cluster state is invalid")
}
// get node // get node
node, ok := cluster.Nodes[nodeName] node, ok := cluster.Nodes[nodeName]
if !ok { if !ok {
@@ -149,15 +153,15 @@ func (cluster *Cluster) GetNode(nodeName string) (*Node, error) {
} }
func SyncNode(cluster *Cluster, nodeName string) error { func SyncNode(cluster *Cluster, nodeName string) error {
cluster.OK = false
err := cluster.BuildNode(nodeName) err := cluster.BuildNode(nodeName)
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
err = cluster.ResolvePoolMembership() err = cluster.ResolvePoolMembership()
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
@@ -195,7 +199,7 @@ func (cluster *Cluster) BuildNode(nodeName string) error {
for _, vmid := range vms { for _, vmid := range vms {
wg.Go(func() error { wg.Go(func() error {
start := time.Now() start := time.Now()
err := node.BuildInstance(VM, vmid) err := node.BuildInstance(VM, paas.SafeUint64(vmid))
if err != nil { // if an error was encountered, continue and log the error 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) log.Printf("[ERR ] error encountered while syncing vm %s.%d: %s", nodeName, vmid, err)
} else { } else {
@@ -213,7 +217,7 @@ func (cluster *Cluster) BuildNode(nodeName string) error {
for _, vmid := range cts { for _, vmid := range cts {
wg.Go(func() error { wg.Go(func() error {
start := time.Now() start := time.Now()
err := node.BuildInstance(CT, vmid) err := node.BuildInstance(CT, paas.SafeUint64(vmid))
if err != nil { // if an error was encountered, continue and log the error 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) log.Printf("[ERR ] error encountered while syncing ct %s.%d: %s", nodeName, vmid, err)
} else { } else {
@@ -247,7 +251,21 @@ func (cluster *Cluster) BuildNode(nodeName string) error {
return nil return nil
} }
func (node *Node) GetInstance(vmid uint) (*Instance, error) { func (cluster *Cluster) GetInstance(nodeName string, vmid uint64) (*Instance, error) {
// aquire cluster lock
cluster.lock.Lock()
defer cluster.lock.Unlock()
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)
}
// aquire node lock // aquire node lock
node.lock.Lock() node.lock.Lock()
defer node.lock.Unlock() defer node.lock.Unlock()
@@ -265,26 +283,28 @@ func (node *Node) GetInstance(vmid uint) (*Instance, error) {
} }
} }
func SyncInstance(cluster *Cluster, nodeName string, vmid uint) error { func SyncInstance(cluster *Cluster, nodeName string, vmid uint64) error {
cluster.OK = false
node, err := cluster.GetNode(nodeName) node, err := cluster.GetNode(nodeName)
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
instance, err := node.GetInstance(uint(vmid)) instance, err := cluster.GetInstance(nodeName, vmid)
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
err = node.BuildInstance(instance.Type, uint(vmid)) err = node.BuildInstance(instance.Type, vmid)
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
err = cluster.ResolvePoolMembership() err = cluster.ResolvePoolMembership()
if err != nil { if err != nil {
cluster.OK = false
return err return err
} }
@@ -294,7 +314,7 @@ func SyncInstance(cluster *Cluster, nodeName string, vmid uint) error {
// hard sync instance // hard sync instance
// returns error if the instance could not be reached // returns error if the instance could not be reached
func (node *Node) BuildInstance(instancetype InstanceType, vmid uint) error { func (node *Node) BuildInstance(instancetype InstanceType, vmid uint64) error {
instanceID := InstanceID(vmid) instanceID := InstanceID(vmid)
var instance *Instance var instance *Instance
var err error var err error
+2 -2
View File
@@ -174,7 +174,7 @@ 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(VMID uint64) (*Instance, error) {
instance := Instance{} 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 {
@@ -221,7 +221,7 @@ 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(VMID uint64) (*Instance, error) {
instance := Instance{} 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 {