lxd/cluster/heartbeat.go
func Heartbeat(gateway *Gateway, cluster *db.Cluster) (task.Func, task.Schedule) {
	heartbeat := func(ctx context.Context) {
		if gateway.server == nil || gateway.memoryDial != nil {
			// We're not a raft node or we're not clustered
			return
		}

		raftNodes, err := gateway.currentRaftNodes()
		if err == raft.ErrNotLeader {
			return
		}
		logger.Debugf("Starting heartbeat round")
		if err != nil {
			logger.Warnf("Failed to get current raft nodes: %v", err)
			return
		}

		// Replace the local raft_nodes table immediately because it
		// might miss a row containing ourselves, since we might have
		// been elected leader before the former leader had chance to
		// send us a fresh update through the heartbeat pool.
		logger.Debugf("Heartbeat updating local raft nodes to %+v", raftNodes)
		err = gateway.db.Transaction(func(tx *db.NodeTx) error {
			return tx.RaftNodesReplace(raftNodes)
		})
		if err != nil {
			logger.Warnf("Failed to replace local raft nodes: %v", err)
			return
		}

		var nodes []db.NodeInfo
		var nodeAddress string // Address of this node
		err = cluster.Transaction(func(tx *db.ClusterTx) error {
			var err error
			nodes, err = tx.Nodes()
			if err != nil {
				return err
			}
			nodeAddress, err = tx.NodeAddress()
			if err != nil {
				return err
			}
			return nil
		})
		if err != nil {
			logger.Warnf("Failed to get current cluster nodes: %v", err)
			return
		}

		heartbeats := make([]time.Time, len(nodes))
		heartbeatsLock := sync.Mutex{}
		heartbeatsWg := sync.WaitGroup{}

		for i, node := range nodes {
			// Special case the local node
			if node.Address == nodeAddress {
				heartbeatsLock.Lock()
				heartbeats[i] = time.Now()
				heartbeatsLock.Unlock()
				continue
			}

			// Parallelize the rest
			heartbeatsWg.Add(1)
			go func(i int, address string) {
				defer heartbeatsWg.Done()

				// Spread in time by waiting up to 3s less than the interval
				time.Sleep(time.Duration(rand.Intn((heartbeatInterval*1000)-3000)) * time.Millisecond)
				logger.Debugf("Sending heartbeat to %s", address)

				err := heartbeatNode(ctx, address, gateway.cert, raftNodes)
				if err == nil {
					heartbeatsLock.Lock()
					heartbeats[i] = time.Now()
					heartbeatsLock.Unlock()
					logger.Debugf("Successful heartbeat for %s", address)
				} else {
					logger.Debugf("Failed heartbeat for %s: %v", address, err)
				}
			}(i, node.Address)
		}
		heartbeatsWg.Wait()

		// If the context has been cancelled, return immediately.
		if ctx.Err() != nil {
			logger.Debugf("Aborting heartbeat round")
			return
		}

		err = cluster.Transaction(func(tx *db.ClusterTx) error {
			for i, node := range nodes {
				if heartbeats[i].Equal(time.Time{}) {
					continue
				}

				err := tx.NodeHeartbeat(node.Address, heartbeats[i])
				if err != nil {
					return err
				}
			}
			return nil
		})
		if err != nil {
			logger.Warnf("Failed to update heartbeat: %v", err)
		}
		logger.Debugf("Completed heartbeat round")
	}

	// Since the database APIs are blocking we need to wrap the core logic
	// and run it in a goroutine, so we can abort as soon as the context expires.
	heartbeatWrapper := func(ctx context.Context) {
		ch := make(chan struct{})
		go func() {
			heartbeat(ctx)
			ch <- struct{}{}
		}()
		select {
		case <-ch:
		case <-ctx.Done():
		}
	}

	schedule := task.Every(time.Duration(heartbeatInterval) * time.Second)

	return heartbeatWrapper, schedule
}
func Heartbeat(gateway *Gateway, cluster *db.Cluster) (task.Func, task.Schedule) {
	heartbeat := func(ctx context.Context) {
		if gateway.server == nil || gateway.memoryDial != nil {
			// We're not a raft node or we're not clustered
			return
		}

		raftNodes, err := gateway.currentRaftNodes()
		if err == raft.ErrNotLeader {
			return
		}
		logger.Debugf("Starting heartbeat round")
		if err != nil {
			logger.Warnf("Failed to get current raft nodes: %v", err)
			return
		}

		// Replace the local raft_nodes table immediately because it
		// might miss a row containing ourselves, since we might have
		// been elected leader before the former leader had chance to
		// send us a fresh update through the heartbeat pool.
		logger.Debugf("Heartbeat updating local raft nodes to %+v", raftNodes)
		err = gateway.db.Transaction(func(tx *db.NodeTx) error {
			return tx.RaftNodesReplace(raftNodes)
		})
		if err != nil {
			logger.Warnf("Failed to replace local raft nodes: %v", err)
			return
		}

		var nodes []db.NodeInfo
		var nodeAddress string // Address of this node
		err = cluster.Transaction(func(tx *db.ClusterTx) error {
			var err error
			nodes, err = tx.Nodes()
			if err != nil {
				return err
			}
			nodeAddress, err = tx.NodeAddress()
			if err != nil {
				return err
			}
			return nil
		})
		if err != nil {
			logger.Warnf("Failed to get current cluster nodes: %v", err)
			return
		}

		heartbeats := make([]time.Time, len(nodes))
		heartbeatsLock := sync.Mutex{}
		heartbeatsWg := sync.WaitGroup{}

		for i, node := range nodes {
			// Special case the local node
			if node.Address == nodeAddress {
				heartbeatsLock.Lock()
				heartbeats[i] = time.Now()
				heartbeatsLock.Unlock()
				continue
			}

			// Parallelize the rest
			heartbeatsWg.Add(1)
			go func(i int, address string) {
				defer heartbeatsWg.Done()

				// Spread in time by waiting up to 3s less than the interval
				time.Sleep(time.Duration(rand.Intn((heartbeatInterval*1000)-3000)) * time.Millisecond)
				logger.Debugf("Sending heartbeat to %s", address)

				err := heartbeatNode(ctx, address, gateway.cert, raftNodes)
				if err == nil {
					heartbeatsLock.Lock()
					heartbeats[i] = time.Now()
					heartbeatsLock.Unlock()
					logger.Debugf("Successful heartbeat for %s", address)
				} else {
					logger.Debugf("Failed heartbeat for %s: %v", address, err)
				}
			}(i, node.Address)
		}
		heartbeatsWg.Wait()

		// If the context has been cancelled, return immediately.
		if ctx.Err() != nil {
			logger.Debugf("Aborting heartbeat round")
			return
		}

		err = cluster.Transaction(func(tx *db.ClusterTx) error {
			for i, node := range nodes {
				if heartbeats[i].Equal(time.Time{}) {
					continue
				}

				err := tx.NodeHeartbeat(node.Address, heartbeats[i])
				if err != nil {
					return err
				}
			}
			return nil
		})
		if err != nil {
			logger.Warnf("Failed to update heartbeat: %v", err)
		}
		logger.Debugf("Completed heartbeat round")
	}

	// Since the database APIs are blocking we need to wrap the core logic
	// and run it in a goroutine, so we can abort as soon as the context expires.
	heartbeatWrapper := func(ctx context.Context) {
		ch := make(chan struct{})
		go func() {
			heartbeat(ctx)
			ch <- struct{}{}
		}()
		select {
		case <-ch:
		case <-ctx.Done():
		}
	}

	schedule := task.Every(time.Duration(heartbeatInterval) * time.Second)

	return heartbeatWrapper, schedule
}
func Heartbeat ( gateway * Gateway , cluster * db . Cluster ) ( task . Func , task . Schedule ) { heartbeat := func ( ctx context . Context ) { if gateway . server == nil || gateway . memoryDial != nil { // We're not a raft node or we're not clustered return } raftNodes , err := gateway . currentRaftNodes ( ) if err == raft . ErrNotLeader { return } logger . Debugf ( " " ) if err != nil { logger . Warnf ( " " , err ) return } // Replace the local raft_nodes table immediately because it // might miss a row containing ourselves, since we might have // been elected leader before the former leader had chance to // send us a fresh update through the heartbeat pool. logger . Debugf ( " " , raftNodes ) err = gateway . db . Transaction ( func ( tx * db . NodeTx ) error { return tx . RaftNodesReplace ( raftNodes ) } ) if err != nil { logger . Warnf ( " " , err ) return } var nodes [ ] db . NodeInfo var nodeAddress string // Address of this node err = cluster . Transaction ( func ( tx * db . ClusterTx ) error { var err error nodes , err = tx . Nodes ( ) if err != nil { return err } nodeAddress , err = tx . NodeAddress ( ) if err != nil { return err } return nil } ) if err != nil { logger . Warnf ( " " , err ) return } heartbeats := make ( [ ] time . Time , len ( nodes ) ) heartbeatsLock := sync . Mutex { } heartbeatsWg := sync . WaitGroup { } for i , node := range nodes { // Special case the local node if node . Address == nodeAddress { heartbeatsLock . Lock ( ) heartbeats [ i ] = time . Now ( ) heartbeatsLock . Unlock ( ) continue } // Parallelize the rest heartbeatsWg . Add ( 1 ) go func ( i int , address string ) { defer heartbeatsWg . Done ( ) // Spread in time by waiting up to 3s less than the interval time . Sleep ( time . Duration ( rand . Intn ( ( heartbeatInterval * 1000 ) - 3000 ) ) * time . Millisecond ) logger . Debugf ( " " , address ) err := heartbeatNode ( ctx , address , gateway . cert , raftNodes ) if err == nil { heartbeatsLock . Lock ( ) heartbeats [ i ] = time . Now ( ) heartbeatsLock . Unlock ( ) logger . Debugf ( " " , address ) } else { logger . Debugf ( " " , address , err ) } } ( i , node . Address ) } heartbeatsWg . Wait ( ) // If the context has been cancelled, return immediately. if ctx . Err ( ) != nil { logger . Debugf ( " " ) return } err = cluster . Transaction ( func ( tx * db . ClusterTx ) error { for i , node := range nodes { if heartbeats [ i ] . Equal ( time . Time { } ) { continue } err := tx . NodeHeartbeat ( node . Address , heartbeats [ i ] ) if err != nil { return err } } return nil } ) if err != nil { logger . Warnf ( " " , err ) } logger . Debugf ( " " ) } // Since the database APIs are blocking we need to wrap the core logic // and run it in a goroutine, so we can abort as soon as the context expires. heartbeatWrapper := func ( ctx context . Context ) { ch := make ( chan struct { } ) go func ( ) { heartbeat ( ctx ) ch <- struct { } { } } ( ) select { case <- ch : case <- ctx . Done ( ) : } } schedule := task . Every ( time . Duration ( heartbeatInterval ) * time . Second ) return heartbeatWrapper , schedule }
// Heartbeat returns a task function that performs leader-initiated heartbeat
// checks against all LXD nodes in the cluster.
//
// It will update the heartbeat timestamp column of the nodes table
// accordingly, and also notify them of the current list of database nodes.
Heartbeat returns a task function that performs leader - initiated heartbeat checks against all LXD nodes in the cluster . It will update the heartbeat timestamp column of the nodes table accordingly and also notify them of the current list of database nodes .