Change cluster MemberEvents to only be published on convergence. See #2692
Conflicts: akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala akka-cluster/src/main/scala/akka/cluster/ClusterJmx.scala akka-cluster/src/main/scala/akka/cluster/ClusterMetricsCollector.scala akka-cluster/src/main/scala/akka/cluster/ClusterReadView.scala akka-cluster/src/multi-jvm/scala/akka/cluster/MultiNodeClusterSpec.scala akka-docs/rst/cluster/cluster-usage-java.rst akka-docs/rst/cluster/cluster-usage-scala.rst akka-kernel/src/main/dist/bin/akka-cluster
This commit is contained in:
Björn Antonsson
parent
0c9ad2f791
commit
a03460329d
21 changed files with 327 additions and 304 deletions
|
|
@ -5,6 +5,7 @@ package akka.cluster
|
|||
|
||||
import language.postfixOps
|
||||
import scala.collection.immutable
|
||||
import scala.collection.immutable.{ VectorBuilder, SortedSet }
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef, Address }
|
||||
import akka.cluster.ClusterEvent._
|
||||
import akka.cluster.MemberStatus._
|
||||
|
|
@ -33,7 +34,6 @@ object ClusterEvent {
|
|||
case class CurrentClusterState(
|
||||
members: immutable.SortedSet[Member] = immutable.SortedSet.empty,
|
||||
unreachable: Set[Member] = Set.empty,
|
||||
convergence: Boolean = false,
|
||||
seenBy: Set[Address] = Set.empty,
|
||||
leader: Option[Address] = None) extends ClusterDomainEvent {
|
||||
|
||||
|
|
@ -75,57 +75,47 @@ object ClusterEvent {
|
|||
}
|
||||
|
||||
/**
|
||||
* A new member joined the cluster.
|
||||
* A new member joined the cluster. Only published after convergence.
|
||||
*/
|
||||
case class MemberJoined(member: Member) extends MemberEvent {
|
||||
if (member.status != Joining) throw new IllegalArgumentException("Expected Joining status, got: " + member)
|
||||
}
|
||||
|
||||
/**
|
||||
* Member status changed to Up
|
||||
* Member status changed to Up. Only published after convergence.
|
||||
*/
|
||||
case class MemberUp(member: Member) extends MemberEvent {
|
||||
if (member.status != Up) throw new IllegalArgumentException("Expected Up status, got: " + member)
|
||||
}
|
||||
|
||||
/**
|
||||
* Member status changed to Leaving
|
||||
* Member status changed to Leaving. Only published after convergence.
|
||||
*/
|
||||
case class MemberLeft(member: Member) extends MemberEvent {
|
||||
if (member.status != Leaving) throw new IllegalArgumentException("Expected Leaving status, got: " + member)
|
||||
}
|
||||
|
||||
/**
|
||||
* Member status changed to Exiting
|
||||
* Member status changed to Exiting. Only published after convergence.
|
||||
*/
|
||||
case class MemberExited(member: Member) extends MemberEvent {
|
||||
if (member.status != Exiting) throw new IllegalArgumentException("Expected Exiting status, got: " + member)
|
||||
}
|
||||
|
||||
/**
|
||||
* A member is considered as unreachable by the failure detector.
|
||||
*/
|
||||
case class MemberUnreachable(member: Member) extends MemberEvent
|
||||
|
||||
/**
|
||||
* Member status changed to Down
|
||||
* Member status changed to Down. Only published after convergence.
|
||||
*/
|
||||
case class MemberDowned(member: Member) extends MemberEvent {
|
||||
if (member.status != Down) throw new IllegalArgumentException("Expected Down status, got: " + member)
|
||||
}
|
||||
|
||||
/**
|
||||
* Member completely removed from the cluster
|
||||
* Member completely removed from the cluster. Only published after convergence.
|
||||
*/
|
||||
case class MemberRemoved(member: Member) extends MemberEvent {
|
||||
if (member.status != Removed) throw new IllegalArgumentException("Expected Removed status, got: " + member)
|
||||
}
|
||||
|
||||
/**
|
||||
* Cluster convergence state changed.
|
||||
*/
|
||||
case class ConvergenceChanged(convergence: Boolean) extends ClusterDomainEvent
|
||||
|
||||
/**
|
||||
* Leader of the cluster members changed. Only published after convergence.
|
||||
*/
|
||||
|
|
@ -138,6 +128,12 @@ object ClusterEvent {
|
|||
}
|
||||
|
||||
/**
|
||||
* A member is considered as unreachable by the failure detector.
|
||||
*/
|
||||
case class UnreachableMember(member: Member) extends ClusterDomainEvent
|
||||
|
||||
/**
|
||||
* INTERNAL API
|
||||
*
|
||||
* Current snapshot of cluster node metrics. Published to subscribers.
|
||||
*/
|
||||
|
|
@ -163,56 +159,75 @@ object ClusterEvent {
|
|||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
private[cluster] def diff(oldGossip: Gossip, newGossip: Gossip): immutable.IndexedSeq[ClusterDomainEvent] = {
|
||||
val newMembers = newGossip.members -- oldGossip.members
|
||||
private[cluster] def diffUnreachable(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[UnreachableMember] =
|
||||
if (newGossip eq oldGossip) Nil
|
||||
else {
|
||||
val newUnreachable = newGossip.overview.unreachable -- oldGossip.overview.unreachable
|
||||
val unreachableEvents = newUnreachable map UnreachableMember
|
||||
|
||||
val membersGroupedByAddress = (newGossip.members.toList ++ oldGossip.members.toList).groupBy(_.address)
|
||||
val changedMembers = membersGroupedByAddress collect {
|
||||
case (_, newMember :: oldMember :: Nil) if newMember.status != oldMember.status ⇒ newMember
|
||||
immutable.Seq.empty ++ unreachableEvents
|
||||
}
|
||||
|
||||
val memberEvents = (newMembers ++ changedMembers) map { m ⇒
|
||||
if (m.status == Joining) MemberJoined(m)
|
||||
else if (m.status == Up) MemberUp(m)
|
||||
else if (m.status == Leaving) MemberLeft(m)
|
||||
else if (m.status == Exiting) MemberExited(m)
|
||||
else throw new IllegalStateException("Unexpected member status: " + m)
|
||||
/**
|
||||
* INTERNAL API.
|
||||
*/
|
||||
private[cluster] def diffMemberEvents(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[MemberEvent] =
|
||||
if (newGossip eq oldGossip) Nil
|
||||
else {
|
||||
val newMembers = newGossip.members -- oldGossip.members
|
||||
val membersGroupedByAddress = List(newGossip.members, oldGossip.members).flatten.groupBy(_.address)
|
||||
val changedMembers = membersGroupedByAddress collect {
|
||||
case (_, newMember :: oldMember :: Nil) if newMember.status != oldMember.status ⇒ newMember
|
||||
}
|
||||
val memberEvents = (newMembers ++ changedMembers) map { m ⇒
|
||||
m.status match {
|
||||
case Joining ⇒ MemberJoined(m)
|
||||
case Up ⇒ MemberUp(m)
|
||||
case Leaving ⇒ MemberLeft(m)
|
||||
case Exiting ⇒ MemberExited(m)
|
||||
case _ ⇒ throw new IllegalStateException("Unexpected member status: " + m)
|
||||
}
|
||||
}
|
||||
|
||||
val allNewUnreachable = newGossip.overview.unreachable -- oldGossip.overview.unreachable
|
||||
val newDowned = allNewUnreachable filter { _.status == Down }
|
||||
val downedEvents = newDowned map MemberDowned
|
||||
|
||||
val unreachableGroupedByAddress =
|
||||
List(newGossip.overview.unreachable, oldGossip.overview.unreachable).flatten.groupBy(_.address)
|
||||
val unreachableDownMembers = unreachableGroupedByAddress collect {
|
||||
case (_, newMember :: oldMember :: Nil) if newMember.status == Down && newMember.status != oldMember.status ⇒
|
||||
newMember
|
||||
}
|
||||
val unreachableDownedEvents = unreachableDownMembers map MemberDowned
|
||||
|
||||
val removedEvents = (oldGossip.members -- newGossip.members -- newGossip.overview.unreachable) map { m ⇒
|
||||
MemberRemoved(m.copy(status = Removed))
|
||||
}
|
||||
|
||||
(new VectorBuilder[MemberEvent]() ++= memberEvents ++= downedEvents ++= unreachableDownedEvents
|
||||
++= removedEvents).result()
|
||||
}
|
||||
|
||||
val allNewUnreachable = newGossip.overview.unreachable -- oldGossip.overview.unreachable
|
||||
val (newDowned, newUnreachable) = allNewUnreachable partition { _.status == Down }
|
||||
val downedEvents = newDowned map MemberDowned
|
||||
val unreachableEvents = newUnreachable map MemberUnreachable
|
||||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
private[cluster] def diffLeader(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[LeaderChanged] =
|
||||
if (newGossip.leader != oldGossip.leader) List(LeaderChanged(newGossip.leader))
|
||||
else Nil
|
||||
|
||||
val unreachableGroupedByAddress =
|
||||
(newGossip.overview.unreachable.toList ++ oldGossip.overview.unreachable.toList).groupBy(_.address)
|
||||
val unreachableDownMembers = unreachableGroupedByAddress collect {
|
||||
case (_, newMember :: oldMember :: Nil) if newMember.status == Down && newMember.status != oldMember.status ⇒
|
||||
newMember
|
||||
/**
|
||||
* INTERNAL API
|
||||
*/
|
||||
private[cluster] def diffSeen(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[SeenChanged] =
|
||||
if (newGossip eq oldGossip) Nil
|
||||
else {
|
||||
val newConvergence = newGossip.convergence
|
||||
val newSeenBy = newGossip.seenBy
|
||||
if (newConvergence != oldGossip.convergence || newSeenBy != oldGossip.seenBy)
|
||||
List(SeenChanged(newConvergence, newSeenBy))
|
||||
else Nil
|
||||
}
|
||||
val unreachableDownedEvents = unreachableDownMembers map MemberDowned
|
||||
|
||||
val removedEvents = (oldGossip.members -- newGossip.members -- newGossip.overview.unreachable) map { m ⇒
|
||||
MemberRemoved(m.copy(status = Removed))
|
||||
}
|
||||
|
||||
val newConvergence = newGossip.convergence
|
||||
val convergenceChanged = newConvergence != oldGossip.convergence
|
||||
val convergenceEvents = if (convergenceChanged) List(ConvergenceChanged(newConvergence)) else EmptyImmutableSeq
|
||||
|
||||
val leaderEvents =
|
||||
if (newGossip.leader != oldGossip.leader) List(LeaderChanged(newGossip.leader))
|
||||
else EmptyImmutableSeq
|
||||
|
||||
val newSeenBy = newGossip.seenBy
|
||||
val seenEvents =
|
||||
if (convergenceChanged || newSeenBy != oldGossip.seenBy) List(SeenChanged(newConvergence, newSeenBy))
|
||||
else EmptyImmutableSeq
|
||||
|
||||
memberEvents.toVector ++ unreachableEvents ++ downedEvents ++ unreachableDownedEvents ++ removedEvents ++
|
||||
leaderEvents ++ convergenceEvents ++ seenEvents
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -224,34 +239,30 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
|
|||
import InternalClusterAction._
|
||||
|
||||
var latestGossip: Gossip = Gossip()
|
||||
|
||||
// Keep track of LeaderChanged event. Should not be published until
|
||||
// convergence, and it should only be published when leader actually
|
||||
// changed to another node. 3 states:
|
||||
// - None: No LeaderChanged detected yet, nothing published yet
|
||||
// - Some(Left): Stashed LeaderChanged to be published later, when convergence
|
||||
// - Some(Right): Latest published LeaderChanged
|
||||
var leaderChangedState: Option[Either[LeaderChanged, LeaderChanged]] = None
|
||||
var latestConvergedGossip: Gossip = Gossip()
|
||||
var memberEvents: immutable.Seq[MemberEvent] = immutable.Seq.empty
|
||||
|
||||
def receive = {
|
||||
case PublishChanges(oldGossip, newGossip) ⇒ publishChanges(oldGossip, newGossip)
|
||||
case PublishChanges(newGossip) ⇒ publishChanges(newGossip)
|
||||
case currentStats: CurrentInternalStats ⇒ publishInternalStats(currentStats)
|
||||
case PublishCurrentClusterState(receiver) ⇒ publishCurrentClusterState(receiver)
|
||||
case Subscribe(subscriber, to) ⇒ subscribe(subscriber, to)
|
||||
case Unsubscribe(subscriber, to) ⇒ unsubscribe(subscriber, to)
|
||||
case PublishEvent(event) ⇒ publish(event)
|
||||
case PublishDone ⇒ sender ! PublishDone
|
||||
case PublishStart ⇒ publishStart()
|
||||
case PublishDone ⇒ publishDone(sender)
|
||||
}
|
||||
|
||||
def eventStream: EventStream = context.system.eventStream
|
||||
|
||||
def publishCurrentClusterState(receiver: Option[ActorRef]): Unit = {
|
||||
// The state is a mix of converged and latest gossip to mimic what you
|
||||
// would have seen if you where listening to the events.
|
||||
val state = CurrentClusterState(
|
||||
members = latestGossip.members,
|
||||
members = latestConvergedGossip.members,
|
||||
unreachable = latestGossip.overview.unreachable,
|
||||
convergence = latestGossip.convergence,
|
||||
seenBy = latestGossip.seenBy,
|
||||
leader = latestGossip.leader)
|
||||
leader = latestConvergedGossip.leader)
|
||||
receiver match {
|
||||
case Some(ref) ⇒ ref ! state
|
||||
case None ⇒ publish(state)
|
||||
|
|
@ -268,54 +279,43 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
|
|||
case Some(c) ⇒ eventStream.unsubscribe(subscriber, c)
|
||||
}
|
||||
|
||||
def publishChanges(oldGossip: Gossip, newGossip: Gossip): Unit = {
|
||||
def publishChanges(newGossip: Gossip): Unit = {
|
||||
val oldGossip = latestGossip
|
||||
// keep the latestGossip to be sent to new subscribers
|
||||
latestGossip = newGossip
|
||||
diff(oldGossip, newGossip) foreach { event ⇒
|
||||
event match {
|
||||
case x @ LeaderChanged(_) if leaderChangedState == Some(Right(x)) ⇒
|
||||
// skip, this leader has already been published
|
||||
|
||||
case x @ LeaderChanged(_) if oldGossip.convergence && newGossip.convergence ⇒
|
||||
// leader changed and immediate convergence
|
||||
leaderChangedState = Some(Right(x))
|
||||
publish(x)
|
||||
|
||||
case x: LeaderChanged ⇒
|
||||
// publish later, when convergence
|
||||
leaderChangedState = Some(Left(x))
|
||||
|
||||
case ConvergenceChanged(true) ⇒
|
||||
// now it's convergence, publish eventual stashed LeaderChanged event
|
||||
leaderChangedState match {
|
||||
case Some(Left(x)) ⇒
|
||||
leaderChangedState = Some(Right(x))
|
||||
publish(x)
|
||||
|
||||
case _ ⇒ // nothing stashed
|
||||
}
|
||||
publish(event)
|
||||
|
||||
case MemberDowned(m) ⇒
|
||||
// TODO this case might be collapsed with MemberRemoved, see ticket #2788
|
||||
// but right now we don't change Downed to Removed
|
||||
publish(event)
|
||||
// notify DeathWatch about downed node
|
||||
publish(AddressTerminated(m.address))
|
||||
|
||||
case MemberRemoved(m) ⇒
|
||||
publish(event)
|
||||
// notify DeathWatch about removed node
|
||||
publish(AddressTerminated(m.address))
|
||||
|
||||
case _ ⇒
|
||||
// all other events
|
||||
publish(event)
|
||||
}
|
||||
// first publish the diffUnreachable between the last two gossips
|
||||
diffUnreachable(oldGossip, newGossip) foreach { event ⇒
|
||||
publish(event)
|
||||
// notify DeathWatch about unreachable node
|
||||
publish(AddressTerminated(event.member.address))
|
||||
}
|
||||
// buffer up the MemberEvents waiting for convergence
|
||||
memberEvents ++= diffMemberEvents(oldGossip, newGossip)
|
||||
// if we have convergence then publish the MemberEvents and possibly a LeaderChanged
|
||||
if (newGossip.convergence) {
|
||||
val previousConvergedGossip = latestConvergedGossip
|
||||
latestConvergedGossip = newGossip
|
||||
memberEvents foreach publish
|
||||
memberEvents = immutable.Seq.empty
|
||||
diffLeader(previousConvergedGossip, latestConvergedGossip) foreach publish
|
||||
}
|
||||
// publish internal SeenState for testing purposes
|
||||
diffSeen(oldGossip, newGossip) foreach publish
|
||||
}
|
||||
|
||||
def publishInternalStats(currentStats: CurrentInternalStats): Unit = publish(currentStats)
|
||||
|
||||
def publish(event: AnyRef): Unit = eventStream publish event
|
||||
|
||||
def publishStart(): Unit = clearState()
|
||||
|
||||
def publishDone(receiver: ActorRef): Unit = {
|
||||
clearState()
|
||||
receiver ! PublishDoneFinished
|
||||
}
|
||||
|
||||
def clearState(): Unit = {
|
||||
latestGossip = Gossip()
|
||||
latestConvergedGossip = Gossip()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue