The following pull request was submitted through Github.
It can be accessed and reviewed at: https://github.com/lxc/lxd/pull/5984
This e-mail was sent by the LXC bot, direct replies will not reach the author
unless they happen to be subscribed to this list.
=== Description (from pull-request) ===
raft_nodes and nodes tables do not share the same IDs and it is not safe to assume they will always match.
Signed-off-by: Thomas Parrott <thomas.parr...@canonical.com>
From 7ddd991b56ceee1439ce6aa7b509e7914ac60d62 Mon Sep 17 00:00:00 2001
From: Thomas Parrott <thomas.parr...@canonical.com>
Date: Thu, 18 Jul 2019 09:42:52 +0100
Subject: [PATCH] cluster/heartbeat: Stops using node ID for comparison of
nodes
raft_nodes and nodes tables do not share the same IDs and it is not safe to
assume they will always match.
Signed-off-by: Thomas Parrott <thomas.parr...@canonical.com>
---
lxd/cluster/gateway.go | 4 ++--
lxd/cluster/heartbeat.go | 41 ++++++++++++++++++++--------------------
2 files changed, 22 insertions(+), 23 deletions(-)
diff --git a/lxd/cluster/gateway.go b/lxd/cluster/gateway.go
index 78a9264a46..57e4ba4468 100644
--- a/lxd/cluster/gateway.go
+++ b/lxd/cluster/gateway.go
@@ -179,9 +179,9 @@ func (g *Gateway) HandlerFuncs(nodeRefreshTask
func(*APIHeartbeat)) map[string]h
nodes := make([]db.RaftNode, 0)
for _, node := range heartbeatData.Members {
- if node.Raft {
+ if node.RaftNodeID > 0 {
nodes = append(nodes, db.RaftNode{
- ID: node.ID,
+ ID: node.RaftNodeID,
Address: node.Address,
})
}
diff --git a/lxd/cluster/heartbeat.go b/lxd/cluster/heartbeat.go
index 45bca1d908..4688a5f810 100644
--- a/lxd/cluster/heartbeat.go
+++ b/lxd/cluster/heartbeat.go
@@ -19,12 +19,12 @@ import (
// APIHeartbeatMember contains specific cluster node info.
type APIHeartbeatMember struct {
- ID int64
- Address string
- Raft bool
- LastHeartbeat time.Time
- Online bool // Calculated from offline threshold and
LastHeatbeat time.
- updated bool // Has node been updated during this heartbeat run.
Not sent to nodes.
+ LXDNodeID int64 // The ID field value in the nodes table.
+ RaftNodeID int64 // The ID field value in the raft_nodes table
(zero if non-raft node).
+ Address string // The IP address and port of the node.
+ LastHeartbeat time.Time // The last time a positive heartbeat response
was received.
+ Online bool // Calculated from offline threshold and
LastHeatbeat time.
+ updated bool // Has node been updated during this heartbeat
run. Not sent to nodes.
}
// APIHeartbeatVersion contains max versions for all nodes in cluster.
@@ -36,7 +36,7 @@ type APIHeartbeatVersion struct {
// APIHeartbeat contains data sent to nodes in heartbeat.
type APIHeartbeat struct {
sync.Mutex // Used to control access to Members maps.
- Members map[int64]APIHeartbeatMember
+ Members map[string]APIHeartbeatMember
Version APIHeartbeatVersion
Time time.Time
@@ -53,7 +53,7 @@ func (hbState *APIHeartbeat) Update(fullStateList bool,
raftNodes []db.RaftNode,
hbState.Time = time.Now()
if hbState.Members == nil {
- hbState.Members = make(map[int64]APIHeartbeatMember)
+ hbState.Members = make(map[string]APIHeartbeatMember)
}
// If we've been supplied a fresh set of node states, this is a full
state list.
@@ -61,24 +61,22 @@ func (hbState *APIHeartbeat) Update(fullStateList bool,
raftNodes []db.RaftNode,
// Add raft nodes first with the raft flag set to true, but missing
LastHeartbeat time.
for _, node := range raftNodes {
- member, exists := hbState.Members[node.ID]
+ member, exists := hbState.Members[node.Address]
if !exists {
member = APIHeartbeatMember{
- ID: node.ID,
Address: node.Address,
}
}
- member.Raft = true
- hbState.Members[node.ID] = member
+ member.RaftNodeID = node.ID
+ hbState.Members[node.Address] = member
}
// Add remaining nodes, and when if existing node is found, update
status.
for _, node := range allNodes {
- member, exists := hbState.Members[node.ID]
+ member, exists := hbState.Members[node.Address]
if !exists {
member = APIHeartbeatMember{
- ID: node.ID,
Address: node.Address,
}
}
@@ -87,8 +85,9 @@ func (hbState *APIHeartbeat) Update(fullStateList bool,
raftNodes []db.RaftNode,
member.LastHeartbeat = node.Heartbeat
}
+ member.LXDNodeID = node.ID
member.Online =
!member.LastHeartbeat.Before(time.Now().Add(-offlineThreshold))
- hbState.Members[node.ID] = member
+ hbState.Members[node.Address] = member
// Keep a record of highest APIExtensions and Schema version
seen in all nodes.
if node.APIExtensions > maxAPIExtensionsVersion {
@@ -111,7 +110,7 @@ func (hbState *APIHeartbeat) Update(fullStateList bool,
raftNodes []db.RaftNode,
// Send sends heartbeat requests to the nodes supplied and updates heartbeat
state.
func (hbState *APIHeartbeat) Send(ctx context.Context, cert *shared.CertInfo,
localAddress string, nodes []db.NodeInfo, delay bool) {
heartbeatsWg := sync.WaitGroup{}
- sendHeartbeat := func(nodeID int64, address string, delay bool,
heartbeatData *APIHeartbeat) {
+ sendHeartbeat := func(address string, delay bool, heartbeatData
*APIHeartbeat) {
defer heartbeatsWg.Done()
if delay {
@@ -125,7 +124,7 @@ func (hbState *APIHeartbeat) Send(ctx context.Context, cert
*shared.CertInfo, lo
if err == nil {
hbState.Lock()
// Ensure only update nodes that exist in Members
already.
- hbNode, existing := hbState.Members[nodeID]
+ hbNode, existing := hbState.Members[address]
if !existing {
return
}
@@ -133,7 +132,7 @@ func (hbState *APIHeartbeat) Send(ctx context.Context, cert
*shared.CertInfo, lo
hbNode.LastHeartbeat = time.Now()
hbNode.Online = true
hbNode.updated = true
- hbState.Members[nodeID] = hbNode
+ hbState.Members[address] = hbNode
hbState.Unlock()
logger.Debugf("Successful heartbeat for %s", address)
} else {
@@ -145,18 +144,18 @@ func (hbState *APIHeartbeat) Send(ctx context.Context,
cert *shared.CertInfo, lo
// Special case for the local node - just record the time now.
if node.Address == localAddress {
hbState.Lock()
- hbNode := hbState.Members[node.ID]
+ hbNode := hbState.Members[node.Address]
hbNode.LastHeartbeat = time.Now()
hbNode.Online = true
hbNode.updated = true
- hbState.Members[node.ID] = hbNode
+ hbState.Members[node.Address] = hbNode
hbState.Unlock()
continue
}
// Parallelize the rest.
heartbeatsWg.Add(1)
- go sendHeartbeat(node.ID, node.Address, delay, hbState)
+ go sendHeartbeat(node.Address, delay, hbState)
}
heartbeatsWg.Wait()
}
_______________________________________________
lxc-devel mailing list
lxc-devel@lists.linuxcontainers.org
http://lists.linuxcontainers.org/listinfo/lxc-devel
