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
}
// 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.