=cluster #52 Leader actions on convergence, including .removals #376
Conversation
| import NIOSSL | ||
| import XCTest | ||
|
|
||
| final class ClusterLeaderActionsClusteredTests: ClusteredNodesTestBase { |
There was a problem hiding this comment.
This file was just renamed, the new test is ClusterLeaderActionsTests
|
Well, nice to see this green :) Will give it another skim after the ski trip. |
ktoso
changed the title
|
Time to rebase! |
|
🤞 |
…ore predictable stable actions
…a better impl in DowningClusteredTests now
|
And we're green :-) Fixed the wrongly pushed sample... I realize review probably is hard for this one -- not expecting a full review really. But I'll point out some points that could be worth a look / read. I plan to write cluster documentation which will be more important and easier to follow in terms of what semantics do / we want to offer :-) |
| if settings.cluster.enabled { | ||
| self.log.info("Actor System Settings in effect: Cluster.autoLeaderElection: \(self.settings.cluster.autoLeaderElection)") | ||
| self.log.info("Actor System Settings in effect: Cluster.downingStrategy: \(self.settings.cluster.downingStrategy)") | ||
| } |
There was a problem hiding this comment.
I kind of like this, helps a lot to realize what mode the cluster is in and what to expect of it...
Open to ideas if this is too loud or nice
There was a problem hiding this comment.
Or maybe allow turning it off?
There was a problem hiding this comment.
I think it's nice. Only on system start-up so shouldn't be too noisy.
|
|
||
| case .snapshot(let snapshot): | ||
| Cluster.Membership.diff(from: .empty, to: snapshot).changes.forEach { change in | ||
| Cluster.Membership._diff(from: .empty, to: snapshot).changes.forEach { change in |
There was a problem hiding this comment.
I found diffing to not be super correct, the "change" style APIs are the correct and well tested ones.
So this will be internal kind of.
There was a problem hiding this comment.
We only used it here btw from "empty" so it works fine. Will revisit it though
| var s = self | ||
| s.state.removeValue(forKey: replicaId) | ||
| return s | ||
| } |
There was a problem hiding this comment.
We need this since when we remove members completely from membership we need to also prune the seen tables.
So a table like:
A: A@1 B@5
B: A@1 B@5
when we remove B must become
A: A@1
so we remove the entire vector that B owned, but also prune it from A's vector.
| /// | ||
| /// Used to guarantee phrases like "all nodes have seen a node A in status S", upon which the Leader may act. | ||
| struct Gossip { | ||
| struct Gossip: Equatable { |
| /// The version vector of this gossip and the `Membership` state owned by it. | ||
| var version: VersionVector { | ||
| self.seen.table[self.owner]! // !-safe, since we _always)_ know our own world view | ||
| self.seen.underlying[self.owner]! // !-safe, since we _always_ know our own world view |
There was a problem hiding this comment.
we never remove "us" from a gossip, even merges which "remove" are protected from that, which is why the mergeForward takes myself so we never remove "us" from a membership.
The ONLY place where a .removed member may be stored is on that member itself, other members would remove and tombstone it. We want to store it so we know that "damn, not only am I down. Should not really happen in many cases though
| /// Sequence number at which this node was moved to `.up` by a leader. | ||
| /// The sequence starts at `1`, and 0 means the node was not moved to up _yet_. | ||
| public var upNumber: Int? | ||
| public var _upNumber: Int? |
There was a problem hiding this comment.
TODO: they don't work just yet, but will shortly, This will enable more stable singletons (that move around less in face of nodes dying and leaders moving around)
| case .joining, .up, .leaving: | ||
| return Member(node: self.node, status: .down) | ||
| case .down, .removed: | ||
| return self |
There was a problem hiding this comment.
This is a helper so we don't cause a removed -> down accidentally
| /// Note that moving only happens along the lifecycle of a member, e.g. trying to move forward from .up do .joining | ||
| /// will result in a `nil` change and no changes being made to the member. | ||
| public mutating func moveForward(_ status: Cluster.MemberStatus) -> Cluster.MembershipChange? { | ||
| public mutating func moveForward(to status: Cluster.MemberStatus) -> Cluster.MembershipChange? { |
There was a problem hiding this comment.
function enforces that we only move "forward" in status
There was a problem hiding this comment.
if no change, we return nil.
This is super important (!), since whenever we apply a change from gossip or locally, we get a change -- and if it's NOT nil then we fire Cluster.Event -- people should not be getting events if the from/to is the same :-)
yim-lee
left a comment
yim-lee
left a comment
There was a problem hiding this comment.
Code very nicely commented throughout. 🙇♀
| fatalError("Received fork busy response from an unexpected fork: \(fork)! Already in hand: \(inHand), and pending: \(pending)") | ||
|
|
||
| // Ignore others... | ||
| // Ignore others... |
There was a problem hiding this comment.
Yea... since it kind of is in the "previous case" meh
| } | ||
|
|
||
| /// Prune any trace of the passed in replica id. | ||
| /// |
| case .some(let locallyKnownTo) where locallyKnownTo.status.isDown: | ||
| // we have NOT removed it yet, but it is down, so we ignore it | ||
| return .init(causalRelation: causalRelation, effectiveChanges: []) | ||
| case .none where Cluster.MemberStatus.down <= incomingOwnerMember.status: |
There was a problem hiding this comment.
should this be >=? i.e., incomingOwnerMember is down or worse?
There was a problem hiding this comment.
I think you're right! Thanks! I rewrote as incomingOwnerMember.status.isAtLeastDown
| // TODO: consider doing the same for .same? | ||
| return .init(causalRelation: causalRelation, effectiveChanges: []) | ||
| changes = [] | ||
| // self.seen.merge(selfOwner: self.owner, incoming: incoming) |
| get { | ||
| self._latestGossip | ||
| } | ||
| // --------------------- |
| table.version(at: self.nodeB).shouldEqual(VersionVector.parse("B:1 A:1", nodes: self.allNodes)) | ||
| table.version(at: self.nodeC).shouldBeNil() | ||
| } | ||
| } |
| A: A@5 B@5 C@6 | ||
| B: A@5 B@10 C@6 | ||
| C: A@5 B@5 C@6 | ||
| """, |
There was a problem hiding this comment.
is it weird for me to think this looks cute? 🤣
There was a problem hiding this comment.
I like it a lot as well 😻
These tests were insanity without this DSL, now it's super clear to follow <3
| gossip.seen.version(at: self.nodeB).shouldEqual(expected.seen.version(at: self.nodeB)) | ||
| gossip.seen.version(at: self.nodeC).shouldEqual(expected.seen.version(at: self.nodeC)) | ||
| // gossip.seen.shouldEqual(expected.seen) | ||
| // gossip.membership.shouldEqual(expected.membership) |
There was a problem hiding this comment.
Thansk! was for debugging, sometimes hard to "visually diff" :)
|
|
||
| pinfo("Gossip converged on all \(gossips.count) members, after \(gossipSend) (individual) sends") | ||
| } | ||
| } |
| /// This ordering somewhat unusual, however always consistent and used to select a leader -- see `LowestReachableMember`. | ||
| public static let lowestAddressOrdering: (Cluster.Member, Cluster.Member) -> Bool = { l, r in | ||
| l.node < r.node | ||
| } |
There was a problem hiding this comment.
moved from ad hoc impl into a named one; this is how we select leaders
| /// managed by a leader action, rather than (as .down is) be possible to invoke by any node at any time. | ||
| /// Note, that a removal also ensures storage of tombstones on the networking layer, such that any future attempts | ||
| /// of such node re-connecting will be automatically rejected, disallowing the node to "come back" (which we'd call a "zombie" node). | ||
| case removed |
There was a problem hiding this comment.
note what .removed means :)
|
|
||
| return true | ||
| } | ||
| } |
There was a problem hiding this comment.
This is very on purpose manually implemented, see docs
| /// Compute a diff between two membership states. | ||
| /// The diff includes any member state changes, as well as | ||
| internal static func diff(from: Cluster.Membership, to: Cluster.Membership) -> MembershipDiff { | ||
| // TODO: diffing is not super well tested, may lose up numbers |
| public internal(set) var fromStatus: Cluster.MemberStatus? | ||
| public let toStatus: Cluster.MemberStatus | ||
|
|
||
| init(member: Cluster.Member, toStatus: Cluster.MemberStatus? = nil) { |
|
|
||
| var leadershipActions: [LeaderAction] = [] | ||
| leadershipActions.append(contentsOf: collectMemberUpMoves()) | ||
| leadershipActions.append(contentsOf: collectDownMemberRemovals()) |
There was a problem hiding this comment.
The order of those is important, such that we first signal ups and then removals -- such that if any "i will take over for the now dead node" MAY consider the newly added to up nodes
| get { | ||
| self._latestGossip | ||
| } | ||
| // --------------------- |
| // KEEP VERSION >>> \(newValue.seen.version(at: self.myselfNode)) | ||
| // NOW: \(self._latestGossip) | ||
| // NEW: \(newValue) | ||
| // """) |
| self._latestGossip = newValue | ||
| } else { | ||
| self._latestGossip = newValue.incrementingOwnerVersion() | ||
| precondition("\(self._latestGossip.membership)" != "\(newValue.membership)", "WHY! ARE THOSE EQUAL: \(reflecting: self._latestGossip.membership) ||||| \(reflecting: newValue.membership)") |
| } | ||
|
|
||
| if change.isAtLeastDown || change.isRemoval || change.isReplacement { | ||
| if change.isAtLeastDown { |
| proto = .down | ||
| case .removed: | ||
| proto = .removed | ||
| } |
There was a problem hiding this comment.
We fixed the order of those, so leaving is before down, not after -- adjusted all places where we use it to follow the same order.
| [ | ||
| "swim/membersToPing": "\(self.membersToPing)", | ||
| "swim/protocolPeriod": "\(self.protocolPeriod)", | ||
| "swim/timeoutSuspectsBeforePeriod": "\(self.timeoutSuspectsBeforePeriod)", |
There was a problem hiding this comment.
This helps in debugging, by looking at this number we know which should become unreachable
| remoteControl.sendUserMessage(type: Message.self, envelope: Envelope(payload: .message(message)), recipient: self.address) | ||
| } else { | ||
| pprint("no remote control!!!! \(self.address)") | ||
| self.system.log.warning("[SWIM] No remote control, while sending to: \(self.address)") |
|
|
||
| // TODO: if we have seen tables, we can use them to bias the gossip towards the "more behind" nodes | ||
| context.log.trace("Sending gossip to \(target)", metadata: [ | ||
| context.log.info("Sending gossip to \(target)", metadata: [ |
There was a problem hiding this comment.
TODO make trace again, too loud logs
| p += " Serializer (id:\(id)) key:\(key) = \(self.serializers[id], orElse: "<undefined>")\n" | ||
| } | ||
| print(p) | ||
| self.log.debug("\(p)") |
There was a problem hiding this comment.
so the printout gets captured in tests
| func detailedMessage(got it: Any, unexpectedEqual: Any) -> String { | ||
| let msg = "[\(it)] does equal: [\(unexpectedEqual)]\n" | ||
| return self.detailedMessage(msg) | ||
| } |
There was a problem hiding this comment.
since the message was wrong before for "not" checks
| import XCTest | ||
|
|
||
| // "Get down!" | ||
| final class DowningClusteredTests: ClusteredNodesTestBase { |
There was a problem hiding this comment.
The most important test here 👀
It excercises everything -- leaders moving up, down, removing, leaders moving to other nodes. Failure detectors kicking in, merging .down observations form other nodes etc.
| final class DowningClusteredTests: ClusteredNodesTestBase { | ||
| enum NodeStopMethod { | ||
| case leaveSelf // TODO: eventually this one will be more graceful, ensure others see us leave etc | ||
| case leaveSelfNode // TODO: eventually this one will be more graceful, ensure others see us leave etc |
|
|
||
| // ==== ------------------------------------------------------------------------------------------------------------ | ||
| // MARK: Merging gossips | ||
|
|
There was a problem hiding this comment.
These tests explain the best how merging observations work 👍
| gossip.converged().shouldBeTrue() | ||
| } | ||
|
|
||
| // FIXME: we should not need .joining nodes to participate on convergence() |
…erge Co-Authored-By: Yim Lee <[email protected]>
ktoso
changed the title
|
🎉 🎉 🎉 Now I'm excited to put through some testing using the sensors app, then more and more integration and terrible edge case tests 👍 |
2 participants
Re-approaching the leader actions with unit tests rather than end to end ones first.
Quite WIP still, more progress on monday.
This same style to be done for upNumbers (or however I rephrase them, started on them for "oldest" and disliked it somehow...)
Eventually resolves Move nodes in membership to removed when ready to do so #52
Move nodes in membership to removed when ready to do soResolves Leader actions only on convergence #385 Leader actions only on convergence
Depends on =swim #397 swim must detect unreachable via other nodes #400 and SWIMShell: SWIM shell never emits .reachable after a node is found reachable again #401 SWIM hardening work.