This is an automated email from the ASF dual-hosted git repository.
markusthoemmes pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git
The following commit(s) were added to refs/heads/master by this push:
new 53bdf17 Fix bugs in invoker ids and health protocol bookkeeping.
(#4017)
53bdf17 is described below
commit 53bdf1753cd7b99fd753523c314810b68c79ae1d
Author: Vadim Raskin <[email protected]>
AuthorDate: Fri Sep 21 11:20:13 2018 +0200
Fix bugs in invoker ids and health protocol bookkeeping. (#4017)
---
.../core/loadBalancer/InvokerSupervision.scala | 20 +++++---
.../test/InvokerSupervisionTests.scala | 54 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 6 deletions(-)
diff --git
a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
index 18f5ac3..98e6cd4 100644
---
a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
+++
b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
@@ -113,14 +113,21 @@ class InvokerPool(childFactory: (ActorRefFactory,
InvokerInstanceId) => ActorRef
// State of the actor. Mutable vars with immutable collections prevents
closures or messages
// from leaking the state for external mutation
- var instanceToRef = immutable.Map.empty[InvokerInstanceId, ActorRef]
+ var instanceToRef = immutable.Map.empty[Int, ActorRef]
var refToInstance = immutable.Map.empty[ActorRef, InvokerInstanceId]
var status = IndexedSeq[InvokerHealth]()
def receive: Receive = {
case p: PingMessage =>
- val invoker = instanceToRef.getOrElse(p.instance,
registerInvoker(p.instance))
- instanceToRef = instanceToRef.updated(p.instance, invoker)
+ val invoker = instanceToRef.getOrElse(p.instance.toInt,
registerInvoker(p.instance))
+ instanceToRef = instanceToRef.updated(p.instance.toInt, invoker)
+
+ // For the case when the invoker was restarted and got a new displayed
name
+ val oldHealth = status(p.instance.toInt)
+ if (oldHealth.id != p.instance) {
+ status = status.updated(p.instance.toInt, new
InvokerHealth(p.instance, oldHealth.status))
+ refToInstance = refToInstance.updated(invoker, p.instance)
+ }
invoker.forward(p)
@@ -128,7 +135,7 @@ class InvokerPool(childFactory: (ActorRefFactory,
InvokerInstanceId) => ActorRef
case msg: InvocationFinishedMessage =>
// Forward message to invoker, if InvokerActor exists
- instanceToRef.get(msg.invokerInstance).foreach(_.forward(msg))
+ instanceToRef.get(msg.invokerInstance.toInt).foreach(_.forward(msg))
case CurrentState(invoker, currentState: InvokerState) =>
refToInstance.get(invoker).foreach { instance =>
@@ -185,12 +192,13 @@ class InvokerPool(childFactory: (ActorRefFactory,
InvokerInstanceId) => ActorRef
def registerInvoker(instanceId: InvokerInstanceId): ActorRef = {
logging.info(this, s"registered a new invoker:
invoker${instanceId.toInt}")(TransactionId.invokerHealth)
+ // Grow the underlying status sequence to the size needed to contain the
incoming ping. Dummy values are created
+ // to represent invokers, where ping messages haven't arrived yet
status = padToIndexed(status, instanceId.toInt + 1, i => new
InvokerHealth(InvokerInstanceId(i), Offline))
+ status = status.updated(instanceId.toInt, new InvokerHealth(instanceId,
Offline))
val ref = childFactory(context, instanceId)
-
ref ! SubscribeTransitionCallBack(self) // register for state change events
-
refToInstance = refToInstance.updated(ref, instanceId)
ref
diff --git
a/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
b/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
index c8a8c01..9cff1df 100644
---
a/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
+++
b/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
@@ -310,4 +310,58 @@ class InvokerSupervisionTests
invoker.isTimerActive(InvokerActor.timerName) shouldBe false
}
+
+ it should "initially store invoker status with its full id -
instance/uniqueName/displayedName" in {
+
+ val invoker0 = TestProbe()
+ val children = mutable.Queue(invoker0.ref)
+ val childFactory = (f: ActorRefFactory, instance: InvokerInstanceId) =>
children.dequeue()
+
+ val sendActivationToInvoker = stubFunction[ActivationMessage,
InvokerInstanceId, Future[RecordMetadata]]
+ val supervisor = system.actorOf(InvokerPool.props(childFactory,
sendActivationToInvoker, pC))
+
+ val invokerInstance = InvokerInstanceId(0, Some("10.x.x.x"),
Some("invoker-xyz"))
+
+ within(timeout.duration) {
+
+ val ping = PingMessage(invokerInstance)
+
+ supervisor ! ping
+
+ invoker0.expectMsgType[SubscribeTransitionCallBack]
+ invoker0.expectMsg(ping)
+
+ allStates(supervisor) shouldBe IndexedSeq(new
InvokerHealth(invokerInstance, Offline))
+ }
+ }
+
+ it should "update the invoker instance id after it was restarted" in {
+ val invoker0 = TestProbe()
+ val children = mutable.Queue(invoker0.ref)
+ val childFactory = (f: ActorRefFactory, instance: InvokerInstanceId) =>
children.dequeue()
+
+ val sendActivationToInvoker = stubFunction[ActivationMessage,
InvokerInstanceId, Future[RecordMetadata]]
+ val supervisor = system.actorOf(InvokerPool.props(childFactory,
sendActivationToInvoker, pC))
+
+ val invokerInstance = InvokerInstanceId(0, Some("10.x.x.x"),
Some("invoker-xyz"))
+
+ val invokerAfterRestart = InvokerInstanceId(0, Some("10.x.x.x"),
Some("invoker-zyx"))
+
+ within(timeout.duration) {
+ val ping = PingMessage(invokerInstance)
+
+ supervisor ! ping
+
+ invoker0.expectMsgType[SubscribeTransitionCallBack]
+ invoker0.expectMsg(ping)
+
+ invoker0.send(supervisor, CurrentState(invoker0.ref, Unhealthy))
+
+ val newPing = PingMessage(invokerAfterRestart)
+
+ supervisor ! newPing
+
+ allStates(supervisor) shouldBe IndexedSeq(new
InvokerHealth(invokerAfterRestart, Unhealthy))
+ }
+ }
}
