new mutex model

app/app.go

@@ -34,13 +34,10 @@ func Run() {
 
 	cluster := Cluster{}
 	cluster.Init(client)
-	start := time.Now()
 	log.Printf("[INFO] starting cluster sync\n")
 	err := cluster.Sync()
 	if err != nil {
 		log.Printf("[ERR ] error encountered while syncing cluster: %s", err)
-	} else {
-		log.Printf("[INFO] synced cluster in %d ms\n", time.Since(start).Milliseconds())
 	}
 
 	// set repeating update for full rebuilds
@@ -53,15 +50,11 @@ func Run() {
 			case <-channel:
 				return
 			case <-ticker.C:
-				start := time.Now()
 				log.Printf("[INFO] starting cluster sync\n")
 				err := cluster.Sync()
 				if err != nil {
 					log.Printf("[ERR ] error encountered while syncing cluster: %s", err)
-				} else {
-					log.Printf("[INFO] synced cluster in %d ms", time.Since(start).Milliseconds())
 				}
-
 			}
 		}
 	}()
@@ -116,7 +109,6 @@ func Run() {
 	})
 
 	router.POST("/sync", func(c *gin.Context) {
-		start := time.Now()
 		log.Printf("[INFO] starting cluster sync\n")
 		err := cluster.Sync()
 		if err != nil {
@@ -124,21 +116,26 @@ func Run() {
 			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 			return
 		} else {
-			log.Printf("[INFO] synced cluster in %fs\n", time.Since(start).Seconds())
 			return
 		}
 	})
 
 	router.POST("/nodes/:node/sync", func(c *gin.Context) {
 		nodeName := c.Param("node")
-		start := time.Now()
+
 		log.Printf("[INFO] starting %s sync\n", nodeName)
-		err := SyncNode(&cluster, nodeName)
+
+		node, err := cluster.GetNode(nodeName)
+		if err != nil {
+			log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error())
+			c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+			return
+		}
+
+		err = node.Sync()
 		if err != nil {
 			log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error())
 			c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
-		} else {
-			log.Printf("[INFO] synced %s in %fs\n", nodeName, time.Since(start).Seconds())
 		}
 	})
 
@@ -146,17 +143,24 @@ func Run() {
 		nodeName := c.Param("node")
 		vmid, err := strconv.ParseUint(c.Param("vmid"), 10, 64)
 		if err != nil {
+			log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
 			c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", c.Param("instance"))})
 			return
 		}
-		start := time.Now()
+
 		log.Printf("[INFO] starting %s.%d sync\n", nodeName, vmid)
-		err = SyncInstance(&cluster, nodeName, vmid)
+
+		instance, err := cluster.GetInstance(nodeName, vmid)
+		if err != nil {
+			log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
+			c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+			return
+		}
+
+		err = instance.Sync()
 		if err != nil {
 			log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error())
 			c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
-		} else {
-			log.Printf("[INFO] synced %s in %fs\n", nodeName, time.Since(start).Seconds())
 		}
 	})
 

app/model.go

@@ -28,6 +28,9 @@ func (cluster *Cluster) Get() (*Cluster, error) {
 }
 
 func (cluster *Cluster) Sync() error {
+	cluster.lock.Lock()
+	defer cluster.lock.Unlock()
+
 	err := cluster.Build()
 	if err != nil {
 		cluster.OK = false
@@ -45,10 +48,14 @@ func (cluster *Cluster) Sync() error {
 }
 
 // hard sync cluster
+// caller MUST lock cluster BEFORE entering
 func (cluster *Cluster) Build() error {
-	// aquire lock on cluster, release on return
+	// check to confirm cluster is locked before making any changes
-	cluster.lock.Lock()
+	if !cluster.lock.IsLocked() {
-	defer cluster.lock.Unlock()
+		return fmt.Errorf("cluster was not locked before calling cluster.Sync()")
+	}
+
+	start := time.Now()
 
 	cluster.Nodes = make(map[string]*Node)
 
@@ -57,17 +64,16 @@ func (cluster *Cluster) Build() error {
 	// get all nodes
 	nodes, err := cluster.pve.Nodes()
 	if err != nil {
-		cluster.lock.Unlock()
 		return err
 	}
 
 	// for each node:
 	for _, nodeName := range nodes {
 		wg.Go(func() error {
-			start := time.Now()
-
 			// rebuild node
 			node := Node{}
+			node.lock.Lock()
+
 			node.Name = nodeName
 			node.cluster = cluster
 
@@ -75,14 +81,14 @@ func (cluster *Cluster) Build() error {
 			//err := cluster.BuildNode(nodeName)
 			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)
-			} else {
-				log.Printf("[INFO] synced node %s in %d ms", nodeName, time.Since(start).Milliseconds())
 			}
 
 			cluster.NodesLock.Lock()
 			cluster.Nodes[nodeName] = &node
 			cluster.NodesLock.Unlock()
 
+			node.lock.Unlock()
+
 			return err
 		})
 	}
@@ -93,13 +99,20 @@ func (cluster *Cluster) Build() error {
 		return err
 	}
 
+	log.Printf("[INFO] built cluster in %d ms", time.Since(start).Milliseconds())
+
 	return nil
 }
 
+// resolve membership of instances to pools
+// caller MUST lock cluster BEFORE entering
 func (cluster *Cluster) ResolvePoolMembership() error {
-	// aquire lock on cluster, release on return
+	// check to confirm cluster is locked before making any changes
-	cluster.lock.Lock()
+	if !cluster.lock.IsLocked() {
-	defer cluster.lock.Unlock()
+		return fmt.Errorf("cluster was not locked before calling cluster.ResolvePoolMembership()")
+	}
+
+	start := time.Now()
 
 	//clear existing pool memberships
 	for _, node := range cluster.Nodes {
@@ -137,6 +150,8 @@ func (cluster *Cluster) ResolvePoolMembership() error {
 		}
 	}
 
+	log.Printf("[INFO] resovled cluster instance pool memberships in %d ms", time.Since(start).Milliseconds())
+
 	return nil
 }
 
@@ -163,14 +178,16 @@ func (cluster *Cluster) GetNode(nodeName string) (*Node, error) {
 	}
 }
 
-func SyncNode(cluster *Cluster, nodeName string) error {
+func (node *Node) Sync() error {
-	node, err := cluster.GetNode(nodeName)
+	node.lock.Lock()
-	if err != nil {
+	defer node.lock.Unlock()
-		cluster.OK = false
-		return err
-	}
 
-	err = node.Build()
+	cluster := node.cluster
+
+	cluster.lock.Lock()
+	defer cluster.lock.Unlock()
+
+	err := node.Build()
 	if err != nil {
 		cluster.OK = false
 		return err
@@ -188,20 +205,23 @@ func SyncNode(cluster *Cluster, nodeName string) error {
 
 // hard sync node
 // returns error if the node could not be reached
+// caller MUST lock node BEFORE entering
 func (node *Node) Build() error {
+	// check to confirm cluster is locked before making any changes
+	if !node.lock.IsLocked() {
+		return fmt.Errorf("node was not locked before calling node.Build()")
+	}
+
+	start := time.Now()
+
 	nodeName := node.Name
 	cluster := node.cluster
 
-	n, err := cluster.pve.Node(nodeName)
+	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())
 	}
-	*node = *n // copies lock value, which should be unlocked
-
-	// aquire lock on node, release on return
-	node.lock.Lock()
-	defer node.lock.Unlock()
 
 	wg, _ := errgroup.WithContext(context.Background())
 
@@ -219,10 +239,11 @@ func (node *Node) Build() error {
 	}
 	for _, vmid := range vms {
 		wg.Go(func() error {
-			start := time.Now()
 
 			instanceID := InstanceID(paas.SafeUint64(vmid))
 			instance := Instance{}
+			instance.lock.Lock()
+
 			instance.VMID = instanceID
 			instance.Type = VM
 			instance.node = node
@@ -230,14 +251,14 @@ func (node *Node) Build() error {
 			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)
-			} else {
-				log.Printf("[INFO] synced vm %s.%d in %d ms", nodeName, vmid, time.Since(start).Milliseconds())
 			}
 
 			node.InstancesLock.Lock()
 			node.Instances[instanceID] = &instance
 			node.InstancesLock.Unlock()
 
+			instance.lock.Unlock()
+
 			return nil
 		})
 	}
@@ -249,10 +270,10 @@ func (node *Node) Build() error {
 	}
 	for _, vmid := range cts {
 		wg.Go(func() error {
-			start := time.Now()
-
 			instanceID := InstanceID(paas.SafeUint64(vmid))
 			instance := Instance{}
+			instance.lock.Lock()
+
 			instance.VMID = instanceID
 			instance.Type = CT
 			instance.node = node
@@ -260,15 +281,14 @@ func (node *Node) Build() error {
 			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)
-			} else {
-				log.Printf("[INFO] synced ct %s.%d in %d ms", nodeName, vmid, time.Since(start).Milliseconds())
-
 			}
 
 			node.InstancesLock.Lock()
 			node.Instances[instanceID] = &instance
 			node.InstancesLock.Unlock()
 
+			instance.lock.Unlock()
+
 			return nil
 		})
 	}
@@ -289,9 +309,7 @@ func (node *Node) Build() error {
 
 	node.cluster = cluster
 
-	//cluster.NodesLock.Lock()
+	log.Printf("[INFO] built node %s in %d ms", node.Name, time.Since(start).Milliseconds())
-	//cluster.Nodes[nodeName] = node
-	//cluster.NodesLock.Unlock()
 
 	return nil
 }
@@ -328,14 +346,16 @@ func (cluster *Cluster) GetInstance(nodeName string, vmid uint64) (*Instance, er
 	}
 }
 
-func SyncInstance(cluster *Cluster, nodeName string, vmid uint64) error {
+func (instance *Instance) Sync() error {
-	instance, err := cluster.GetInstance(nodeName, vmid)
+	instance.lock.Lock()
-	if err != nil {
+	defer instance.lock.Unlock()
-		cluster.OK = false
-		return err
-	}
 
-	err = instance.Build()
+	cluster := instance.node.cluster
+
+	cluster.lock.Lock()
+	defer cluster.lock.Unlock()
+
+	err := instance.Build()
 	if err != nil {
 		cluster.OK = false
 		return err
@@ -353,17 +373,24 @@ func SyncInstance(cluster *Cluster, nodeName string, vmid uint64) error {
 
 // hard sync instance
 // returns error if the instance could not be reached
+// caller MUST lock cluster BEFORE entering
 func (instance *Instance) Build() error {
+	// check to confirm cluster is locked before making any changes
+	if !instance.lock.IsLocked() {
+		return fmt.Errorf("instance was not locked before calling instance.Build()")
+	}
+
+	start := time.Now()
+
 	vmid := instance.VMID
 	instancetype := instance.Type
 	node := instance.node
-	var i *Instance
 	var err error
 	switch instancetype {
 	case VM:
-		i, err = node.VirtualMachine(vmid)
+		err = node.VirtualMachine(instance, vmid)
 	case CT:
-		i, err = node.Container(vmid)
+		err = node.Container(instance, vmid)
 
 	}
 
@@ -372,12 +399,6 @@ func (instance *Instance) Build() error {
 		return fmt.Errorf("error retrieving %s.%d: %s", node.Name, vmid, err.Error())
 	}
 
-	*instance = *i // copiues lock vlaue which should be unlocked
-
-	// aquire lock on instance, release on return
-	instance.lock.Lock()
-	defer instance.lock.Unlock()
-
 	wg, _ := errgroup.WithContext(context.Background())
 
 	if err != nil && node.Instances[vmid] != nil { // node is unreachable and did exist previously
@@ -433,6 +454,8 @@ func (instance *Instance) Build() error {
 
 	instance.node = node
 
+	log.Printf("[INFO] built instance %s.%d in %d ms", instance.node.Name, instance.VMID, time.Since(start).Milliseconds())
+
 	return nil
 }
 

app/proxmox.go

@@ -73,8 +73,8 @@ func (pve ProxmoxClient) Nodes() ([]string, error) {
 }
 
 // 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.Instances = make(map[InstanceID]*Instance)
 	host.storage = make(map[string][]*proxmox.StorageContent)
@@ -82,14 +82,14 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) {
 	// get pve node
 	node, err := pve.client.Node(context.Background(), nodeName)
 	if err != nil {
-		return &host, err
+		return err
 	}
 
 	// get pve node devices
 	devices := []PVEDevice{}
 	err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/hardware/pci", nodeName), &devices)
 	if err != nil {
-		return &host, err
+		return err
 	}
 	// populate host devices from pve node devices
 	for _, device := range devices {
@@ -119,7 +119,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) {
 	proctypes := []PVEProctype{}
 	err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/capabilities/qemu/cpu", nodeName), &proctypes)
 	if err != nil {
-		return &host, err
+		return err
 	}
 	// populate host proctypes from pve node proctypes
 	for _, proctype := range proctypes {
@@ -130,7 +130,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) {
 	wg, _ := errgroup.WithContext(context.Background())
 	storages, err := node.Storages(context.Background())
 	if err != nil {
-		return &host, err
+		return err
 	}
 	// populate host storage content from pve node storages
 	for _, storage := range storages {
@@ -147,7 +147,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) {
 	}
 	err = wg.Wait()
 	if err != nil {
-		return &host, err
+		return err
 	}
 
 	// set host basic values
@@ -157,7 +157,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) {
 	host.Swap = node.Swap.Total
 	host.pvenode = node
 
-	return &host, err
+	return err
 }
 
 // Get all VM IDs on specified host
@@ -174,11 +174,10 @@ func (host *Node) VirtualMachines() ([]uint, error) {
 }
 
 // Get a VM's CPU, Memory but does not recursively link Devices, Disks, Drives, Nets
-func (host *Node) VirtualMachine(VMID InstanceID) (*Instance, error) {
+func (host *Node) VirtualMachine(instance *Instance, VMID InstanceID) error {
-	instance := Instance{}
 	vm, err := host.pvenode.VirtualMachine(context.Background(), int(VMID))
 	if err != nil {
-		return &instance, err
+		return err
 	}
 
 	config := vm.VirtualMachineConfig
@@ -198,7 +197,7 @@ func (host *Node) VirtualMachine(VMID InstanceID) (*Instance, error) {
 	instance.Nets = make(map[NetID]*Net)
 	instance.Devices = make(map[DeviceID]*Device)
 
-	return &instance, nil
+	return nil
 }
 
 func MergeVMDisksAndUnused(vmc *proxmox.VirtualMachineConfig) map[string]string {
@@ -221,11 +220,10 @@ func (host *Node) Containers() ([]uint, error) {
 }
 
 // Get a CT's CPU, Memory, Swap but does not recursively link Devices, Disks, Drives, Nets
-func (host *Node) Container(VMID InstanceID) (*Instance, error) {
+func (host *Node) Container(instance *Instance, VMID InstanceID) error {
-	instance := Instance{}
 	ct, err := host.pvenode.Container(context.Background(), int(VMID))
 	if err != nil {
-		return &instance, err
+		return err
 	}
 
 	config := ct.ContainerConfig
@@ -243,7 +241,7 @@ func (host *Node) Container(VMID InstanceID) (*Instance, error) {
 	instance.Volumes = make(map[VolumeID]*Volume)
 	instance.Nets = make(map[NetID]*Net)
 
-	return &instance, nil
+	return nil
 }
 
 func MergeCTDisksAndUnused(cc *proxmox.ContainerConfig) map[string]string {

app/types.go

@@ -14,7 +14,7 @@ import (
 // override Nodes map to new custom Node type
 type Cluster struct {
 	paas.Cluster
-	lock      sync.Mutex
+	lock      MutexWithCheck
 	pve       ProxmoxClient
 	NodesLock sync.Mutex
 	Nodes     map[string]*Node `json:"nodes"`
@@ -24,8 +24,8 @@ type Cluster struct {
 // add mutex and pve api Node object
 // override Instances map to new custom Instance type
 type Node struct {
-	lock sync.Mutex
 	paas.Node
+	lock          MutexWithCheck
 	InstancesLock sync.Mutex               // lock for Instances map
 	Instances     map[InstanceID]*Instance `json:"instances"`
 	pvenode       *proxmox.Node
@@ -41,8 +41,8 @@ const CT InstanceType = paas.CT
 
 // add mutex and various config objects
 type Instance struct {
-	lock sync.Mutex
 	paas.Instance
+	lock           MutexWithCheck
 	VolumesLock    sync.Mutex // lock for Volumes map
 	NetsLock       sync.Mutex // lock for Nets map
 	DevicesLock    sync.Mutex // lock for Devices map
@@ -64,3 +64,32 @@ type Device = paas.Device
 type FunctionID = paas.FunctionID
 type Function = paas.Function
 type BootOrder = paas.BootOrder
+
+type MutexWithCheck struct {
+	ml sync.Mutex
+	v  sync.Mutex
+	l  bool
+}
+
+func (m *MutexWithCheck) Lock() {
+	m.ml.Lock()
+	defer m.ml.Unlock()
+
+	m.v.Lock()
+	m.l = true
+}
+
+func (m *MutexWithCheck) Unlock() {
+	m.ml.Lock()
+	defer m.ml.Unlock()
+
+	m.v.Unlock()
+	m.l = false
+}
+
+func (m *MutexWithCheck) IsLocked() bool {
+	m.ml.Lock()
+	defer m.ml.Unlock()
+
+	return m.l
+}