Cluster heartbeat timings (#26757)

akka-cluster/src/main/mima-filters/2.5.23.backwards.excludes

@@ -0,0 +1,8 @@
+# 26757 add timings to cluster heart beat messages
+ProblemFilters.exclude[MissingTypesProblem]("akka.cluster.ClusterHeartbeatSender$Heartbeat$")
+ProblemFilters.exclude[MissingTypesProblem]("akka.cluster.ClusterHeartbeatSender$HeartbeatRsp$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterHeartbeatSender*")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterHeartbeatReceiver.selfHeartbeatRsp")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterHeartbeatReceiver.this")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.cluster.ClusterHeartbeatSender.heartbeatRsp")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.CrossDcHeartbeatSender.selfHeartbeat")

akka-cluster/src/main/protobuf/ClusterMessages.proto

@@ -97,14 +97,26 @@ message CompatibleConfig {
  ****************************************/
 
 /**
- * Heartbeat
- * Sends an Address
+ * Prior to version 2.5.24
+ * Heartbeat sends an Address
+ * Version 2.5.24 can deserialize this message but does not send it.
  */
+message Heartbeat {
+  required Address from = 1;
+  optional int64 sequenceNr = 2;
+  optional sint64 creationTime = 3;
+}
 
 /**
- * HeartbeatRsp
- * Sends an UniqueAddress
+ * Prior to version 2.5.24
+ * HeartbeatRsp Sends an UniqueAddress
+ * Version 2.5.24 can deserialize this message but does not send it.
  */
+message HeartBeatResponse {
+  required UniqueAddress from = 1;
+  optional int64 sequenceNr = 2;
+  optional int64 creationTime = 3;
+}
 
 /****************************************
  * Cluster Gossip Messages

akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala

@@ -223,7 +223,7 @@ private[cluster] final class ClusterDaemon(joinConfigCompatChecker: JoinConfigCo
         Props(classOf[ClusterCoreSupervisor], joinConfigCompatChecker).withDispatcher(context.props.dispatcher),
         name = "core"))
     context.actorOf(
-      Props[ClusterHeartbeatReceiver].withDispatcher(context.props.dispatcher),
+      ClusterHeartbeatReceiver.props(() => Cluster(context.system)).withDispatcher(context.props.dispatcher),
       name = "heartbeatReceiver")
   }
 

akka-cluster/src/main/scala/akka/cluster/ClusterHeartbeat.scala

@@ -8,8 +8,16 @@ import java.util.concurrent.TimeUnit
 
 import scala.annotation.tailrec
 import scala.collection.immutable
-
-import akka.actor.{ Actor, ActorLogging, ActorPath, ActorSelection, Address, DeadLetterSuppression, RootActorPath }
+import akka.actor.{
+  Actor,
+  ActorLogging,
+  ActorPath,
+  ActorSelection,
+  Address,
+  DeadLetterSuppression,
+  Props,
+  RootActorPath
+}
 import akka.cluster.ClusterEvent._
 import akka.remote.FailureDetectorRegistry
 import akka.remote.HeartbeatMessage
@@ -23,19 +31,20 @@ import akka.util.ccompat._
  */
 @InternalApi
 @ccompatUsedUntil213
-private[cluster] final class ClusterHeartbeatReceiver extends Actor with ActorLogging {
+private[cluster] final class ClusterHeartbeatReceiver(getCluster: () => Cluster) extends Actor with ActorLogging {
   import ClusterHeartbeatSender._
 
   // Important - don't use Cluster(context.system) in constructor because that would
   // cause deadlock. See startup sequence in ClusterDaemon.
-  lazy val cluster = Cluster(context.system)
-  lazy val selfHeartbeatRsp = HeartbeatRsp(cluster.selfUniqueAddress)
+  lazy val cluster = getCluster()
+
   lazy val verboseHeartbeat = cluster.settings.Debug.VerboseHeartbeatLogging
 
-  def receive = {
-    case Heartbeat(from) =>
-      if (verboseHeartbeat) cluster.ClusterLogger.logDebug("Heartbeat from [{}]", from)
-      sender() ! selfHeartbeatRsp
+  def receive: Receive = {
+    case hb: Heartbeat =>
+      // TODO log the sequence nr once serializer is enabled
+      if (verboseHeartbeat) cluster.ClusterLogger.logDebug("Heartbeat from [{}]", hb.from)
+      sender() ! HeartbeatRsp(cluster.selfUniqueAddress, hb.sequenceNr, hb.creationTimeNanos)
   }
 
 }
@@ -44,6 +53,7 @@ private[cluster] final class ClusterHeartbeatReceiver extends Actor with ActorLo
 @InternalApi
 private[cluster] object ClusterHeartbeatReceiver {
 
+  def props(clusterFactory: () => Cluster): Props = Props(new ClusterHeartbeatReceiver(clusterFactory))
   def name: String = "heartbeatReceiver"
   def path(address: Address): ActorPath =
     RootActorPath(address) / "system" / "cluster" / name
@@ -57,12 +67,15 @@ private[cluster] object ClusterHeartbeatSender {
   /**
    * Sent at regular intervals for failure detection.
    */
-  final case class Heartbeat(from: Address) extends ClusterMessage with HeartbeatMessage with DeadLetterSuppression
+  final case class Heartbeat(from: Address, sequenceNr: Long, creationTimeNanos: Long)
+      extends ClusterMessage
+      with HeartbeatMessage
+      with DeadLetterSuppression
 
   /**
    * Sent as reply to [[Heartbeat]] messages.
    */
-  final case class HeartbeatRsp(from: UniqueAddress)
+  final case class HeartbeatRsp(from: UniqueAddress, sequenceNr: Long, creationTimeNanos: Long)
       extends ClusterMessage
       with HeartbeatMessage
       with DeadLetterSuppression
@@ -80,7 +93,7 @@ private[cluster] object ClusterHeartbeatSender {
  * a few other nodes, which will reply and then this actor updates the
  * failure detector.
  */
-private[cluster] final class ClusterHeartbeatSender extends Actor with ActorLogging {
+private[cluster] class ClusterHeartbeatSender extends Actor with ActorLogging {
   import ClusterHeartbeatSender._
 
   val cluster = Cluster(context.system)
@@ -93,7 +106,12 @@ private[cluster] final class ClusterHeartbeatSender extends Actor with ActorLogg
   val filterInternalClusterMembers: Member => Boolean =
     _.dataCenter == cluster.selfDataCenter
 
-  val selfHeartbeat = Heartbeat(selfAddress)
+  var sequenceNr = 0
+
+  def selfHeartbeat(): Heartbeat = {
+    sequenceNr += 1
+    Heartbeat(selfAddress, sequenceNr, System.nanoTime())
+  }
 
   val failureDetector = cluster.failureDetector
 
@@ -137,7 +155,7 @@ private[cluster] final class ClusterHeartbeatSender extends Actor with ActorLogg
 
   def active: Actor.Receive = {
     case HeartbeatTick                => heartbeat()
-    case HeartbeatRsp(from)           => heartbeatRsp(from)
+    case response: HeartbeatRsp       => heartbeatRsp(response)
     case MemberRemoved(m, _)          => removeMember(m)
     case evt: MemberEvent             => addMember(evt.member)
     case UnreachableMember(m)         => unreachableMember(m)
@@ -177,6 +195,7 @@ private[cluster] final class ClusterHeartbeatSender extends Actor with ActorLogg
     state = state.reachableMember(m.uniqueAddress)
 
   def heartbeat(): Unit = {
+    val nextHB = selfHeartbeat()
     state.activeReceivers.foreach { to =>
       if (failureDetector.isMonitoring(to.address)) {
         if (verboseHeartbeat) logDebug("Heartbeat to [{}]", to.address)
@@ -186,7 +205,7 @@ private[cluster] final class ClusterHeartbeatSender extends Actor with ActorLogg
         // other side a chance to reply, and also trigger some resends if needed
         scheduler.scheduleOnce(HeartbeatExpectedResponseAfter, self, ExpectedFirstHeartbeat(to))
       }
-      heartbeatReceiver(to.address) ! selfHeartbeat
+      heartbeatReceiver(to.address) ! nextHB
     }
 
     checkTickInterval()
@@ -206,9 +225,10 @@ private[cluster] final class ClusterHeartbeatSender extends Actor with ActorLogg
 
   }
 
-  def heartbeatRsp(from: UniqueAddress): Unit = {
-    if (verboseHeartbeat) logDebug("Heartbeat response from [{}]", from.address)
-    state = state.heartbeatRsp(from)
+  def heartbeatRsp(response: HeartbeatRsp): Unit = {
+    // TODO: log response time and validate sequence nrs once serialisation of sendTime is released
+    if (verboseHeartbeat) logDebug("Heartbeat response from [{}]", response.from.address)
+    state = state.heartbeatRsp(response.from)
   }
 
   def triggerFirstHeartbeat(from: UniqueAddress): Unit =

akka-cluster/src/main/scala/akka/cluster/CrossDcClusterHeartbeat.scala

@@ -34,7 +34,7 @@ import scala.collection.immutable
  */
 @InternalApi
 @ccompatUsedUntil213
-private[cluster] final class CrossDcHeartbeatSender extends Actor with ActorLogging {
+private[cluster] class CrossDcHeartbeatSender extends Actor with ActorLogging {
   import CrossDcHeartbeatSender._
 
   val cluster = Cluster(context.system)
@@ -56,7 +56,12 @@ private[cluster] final class CrossDcHeartbeatSender extends Actor with ActorLogg
 
   val crossDcFailureDetector = cluster.crossDcFailureDetector
 
-  val selfHeartbeat = ClusterHeartbeatSender.Heartbeat(selfAddress)
+  var sequenceNr: Long = 0
+
+  def nextHeartBeat() = {
+    sequenceNr += 1
+    ClusterHeartbeatSender.Heartbeat(selfAddress, sequenceNr, System.nanoTime())
+  }
 
   var dataCentersState: CrossDcHeartbeatingState = CrossDcHeartbeatingState.init(
     selfDataCenter,
@@ -110,7 +115,7 @@ private[cluster] final class CrossDcHeartbeatSender extends Actor with ActorLogg
 
   def active: Actor.Receive = {
     case ClusterHeartbeatSender.HeartbeatTick                => heartbeat()
-    case ClusterHeartbeatSender.HeartbeatRsp(from)           => heartbeatRsp(from)
+    case ClusterHeartbeatSender.HeartbeatRsp(from, _, _)     => heartbeatRsp(from)
     case MemberRemoved(m, _)                                 => removeMember(m)
     case evt: MemberEvent                                    => addMember(evt.member)
     case ClusterHeartbeatSender.ExpectedFirstHeartbeat(from) => triggerFirstHeartbeat(from)
@@ -153,6 +158,7 @@ private[cluster] final class CrossDcHeartbeatSender extends Actor with ActorLogg
     }
 
   def heartbeat(): Unit = {
+    val nextHB = nextHeartBeat()
     dataCentersState.activeReceivers.foreach { to =>
       if (crossDcFailureDetector.isMonitoring(to.address)) {
         if (verboseHeartbeat) logDebug("(Cross) Heartbeat to [{}]", to.address)
@@ -162,7 +168,7 @@ private[cluster] final class CrossDcHeartbeatSender extends Actor with ActorLogg
         // other side a chance to reply, and also trigger some resends if needed
         scheduler.scheduleOnce(HeartbeatExpectedResponseAfter, self, ClusterHeartbeatSender.ExpectedFirstHeartbeat(to))
       }
-      heartbeatReceiver(to.address) ! selfHeartbeat
+      heartbeatReceiver(to.address) ! nextHB
     }
   }
 

akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala

@@ -42,8 +42,12 @@ private[akka] object ClusterMessageSerializer {
   val InitJoinManifest = s"akka.cluster.InternalClusterAction$$InitJoin$$"
   val InitJoinAckManifest = s"akka.cluster.InternalClusterAction$$InitJoinAck"
   val InitJoinNackManifest = s"akka.cluster.InternalClusterAction$$InitJoinNack"
-  val HeartBeatManifest = s"akka.cluster.ClusterHeartbeatSender$$Heartbeat"
-  val HeartBeatRspManifest = s"akka.cluster.ClusterHeartbeatSender$$HeartbeatRsp"
+  // FIXME, remove in a later version (2.6?) and make 2.5.24+ a mandatory step for rolling upgrade
+  val HeartBeatManifestPre2523 = s"akka.cluster.ClusterHeartbeatSender$$Heartbeat"
+  val HeartBeatRspManifest2523 = s"akka.cluster.ClusterHeartbeatSender$$HeartbeatRsp"
+
+  val HeartBeatManifest = "HB"
+  val HeartBeatRspManifest = "HBR"
   val ExitingConfirmedManifest = s"akka.cluster.InternalClusterAction$$ExitingConfirmed"
   val GossipStatusManifest = "akka.cluster.GossipStatus"
   val GossipEnvelopeManifest = "akka.cluster.GossipEnvelope"
@@ -72,8 +76,8 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     case _: InternalClusterAction.InitJoin      => InitJoinManifest
     case _: InternalClusterAction.InitJoinAck   => InitJoinAckManifest
     case _: InternalClusterAction.InitJoinNack  => InitJoinNackManifest
-    case _: ClusterHeartbeatSender.Heartbeat    => HeartBeatManifest
-    case _: ClusterHeartbeatSender.HeartbeatRsp => HeartBeatRspManifest
+    case _: ClusterHeartbeatSender.Heartbeat    => HeartBeatManifestPre2523
+    case _: ClusterHeartbeatSender.HeartbeatRsp => HeartBeatRspManifest2523
     case _: ExitingConfirmed                    => ExitingConfirmedManifest
     case _: GossipStatus                        => GossipStatusManifest
     case _: GossipEnvelope                      => GossipEnvelopeManifest
@@ -83,8 +87,8 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
   }
 
   def toBinary(obj: AnyRef): Array[Byte] = obj match {
-    case ClusterHeartbeatSender.Heartbeat(from)                  => addressToProtoByteArray(from)
-    case ClusterHeartbeatSender.HeartbeatRsp(from)               => uniqueAddressToProtoByteArray(from)
+    case ClusterHeartbeatSender.Heartbeat(from, _, _)            => addressToProtoByteArray(from)
+    case ClusterHeartbeatSender.HeartbeatRsp(from, _, _)         => uniqueAddressToProtoByteArray(from)
     case m: GossipEnvelope                                       => gossipEnvelopeToProto(m).toByteArray
     case m: GossipStatus                                         => gossipStatusToProto(m).toByteArray
     case InternalClusterAction.Join(node, roles)                 => joinToProto(node, roles).toByteArray
@@ -101,8 +105,10 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
   }
 
   def fromBinary(bytes: Array[Byte], manifest: String): AnyRef = manifest match {
+    case HeartBeatManifestPre2523  => deserializeHeartBeatAsAddress(bytes)
+    case HeartBeatRspManifest2523  => deserializeHeartBeatRspAsUniqueAddress(bytes)
     case HeartBeatManifest         => deserializeHeartBeat(bytes)
-    case HeartBeatRspManifest      => deserializeHeartBeatRsp(bytes)
+    case HeartBeatRspManifest      => deserializeHeartBeatResponse(bytes)
     case GossipStatusManifest      => deserializeGossipStatus(bytes)
     case GossipEnvelopeManifest    => deserializeGossipEnvelope(bytes)
     case InitJoinManifest          => deserializeInitJoin(bytes)
@@ -154,9 +160,9 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     case _ => throw new IllegalArgumentException(s"Address [$address] could not be serialized: host or port missing.")
   }
 
-  private def addressToProtoByteArray(address: Address): Array[Byte] = addressToProto(address).build.toByteArray
+  private[akka] def addressToProtoByteArray(address: Address): Array[Byte] = addressToProto(address).build.toByteArray
 
-  private def uniqueAddressToProto(uniqueAddress: UniqueAddress): cm.UniqueAddress.Builder = {
+  private[akka] def uniqueAddressToProto(uniqueAddress: UniqueAddress): cm.UniqueAddress.Builder = {
     cm.UniqueAddress
       .newBuilder()
       .setAddress(addressToProto(uniqueAddress.address))
@@ -277,12 +283,22 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     InternalClusterAction.ExitingConfirmed(uniqueAddressFromBinary(bytes))
   }
 
-  private def deserializeHeartBeatRsp(bytes: Array[Byte]): ClusterHeartbeatSender.HeartbeatRsp = {
-    ClusterHeartbeatSender.HeartbeatRsp(uniqueAddressFromBinary(bytes))
+  private def deserializeHeartBeatRspAsUniqueAddress(bytes: Array[Byte]): ClusterHeartbeatSender.HeartbeatRsp = {
+    ClusterHeartbeatSender.HeartbeatRsp(uniqueAddressFromBinary(bytes), -1, -1)
+  }
+
+  private def deserializeHeartBeatAsAddress(bytes: Array[Byte]): ClusterHeartbeatSender.Heartbeat = {
+    ClusterHeartbeatSender.Heartbeat(addressFromBinary(bytes), -1, -1)
+  }
+
+  def deserializeHeartBeat(bytes: Array[Byte]): ClusterHeartbeatSender.Heartbeat = {
+    val hb = cm.Heartbeat.parseFrom(bytes)
+    ClusterHeartbeatSender.Heartbeat(addressFromProto(hb.getFrom), hb.getSequenceNr, hb.getCreationTime)
   }
 
-  private def deserializeHeartBeat(bytes: Array[Byte]): ClusterHeartbeatSender.Heartbeat = {
-    ClusterHeartbeatSender.Heartbeat(addressFromBinary(bytes))
+  def deserializeHeartBeatResponse(bytes: Array[Byte]): ClusterHeartbeatSender.HeartbeatRsp = {
+    val hbr = cm.HeartBeatResponse.parseFrom(bytes)
+    ClusterHeartbeatSender.HeartbeatRsp(uniqueAddressFromProto(hbr.getFrom), hbr.getSequenceNr, hbr.getCreationTime)
   }
 
   private def deserializeInitJoinNack(bytes: Array[Byte]): InternalClusterAction.InitJoinNack = {

akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatReceiverSpec.scala

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.cluster
+
+import akka.cluster.ClusterHeartbeatSender.{ Heartbeat, HeartbeatRsp }
+import akka.testkit.{ AkkaSpec, ImplicitSender }
+
+class ClusterHeartbeatReceiverSpec extends AkkaSpec("""
+    akka.actor.provider = cluster 
+  """.stripMargin) with ImplicitSender {
+  "ClusterHeartbeatReceiver" should {
+    "respond to heartbeats with the same sequenceNr and sendTime" in {
+      val heartBeater = system.actorOf(ClusterHeartbeatReceiver.props(() => Cluster(system)))
+      heartBeater ! Heartbeat(Cluster(system).selfAddress, 1, 2)
+      expectMsg(HeartbeatRsp(Cluster(system).selfUniqueAddress, 1, 2))
+    }
+  }
+}

akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatSenderSpec.scala

@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.cluster
+
+import akka.actor.{ ActorSelection, Address, Props }
+import akka.cluster.ClusterEvent.{ CurrentClusterState, MemberUp }
+import akka.cluster.ClusterHeartbeatSender.Heartbeat
+import akka.cluster.ClusterHeartbeatSenderSpec.TestClusterHeartBeatSender
+import akka.testkit.{ AkkaSpec, ImplicitSender, TestProbe }
+
+object ClusterHeartbeatSenderSpec {
+  class TestClusterHeartBeatSender(probe: TestProbe) extends ClusterHeartbeatSender {
+    // don't register for cluster events
+    override def preStart(): Unit = {}
+
+    // override where the heart beats go to
+    override def heartbeatReceiver(address: Address): ActorSelection = {
+      context.actorSelection(probe.ref.path)
+    }
+  }
+}
+
+class ClusterHeartbeatSenderSpec extends AkkaSpec("""
+    akka.loglevel = DEBUG
+    akka.actor.provider = cluster 
+    akka.cluster.failure-detector.heartbeat-interval = 0.2s
+  """.stripMargin) with ImplicitSender {
+
+  "ClusterHeartBeatSender" must {
+    "increment heart beat sequence nr" in {
+      val probe = TestProbe()
+      val underTest = system.actorOf(Props(new TestClusterHeartBeatSender(probe)))
+      underTest ! CurrentClusterState()
+      underTest ! MemberUp(
+        Member(UniqueAddress(Address("akka", system.name), 1L), Set("dc-default")).copy(status = MemberStatus.Up))
+
+      probe.expectMsgType[Heartbeat].sequenceNr shouldEqual 1
+      probe.expectMsgType[Heartbeat].sequenceNr shouldEqual 2
+    }
+  }
+
+}