Consider binding action when creating or recovering queue. (apache#5267)

Author: style95

common/scala/src/main/scala/org/apache/openwhisk/core/service/WatcherService.scala

@@ -128,6 +128,7 @@ class WatcherService(etcdClient: EtcdClient)(implicit logging: Logging, actorSys
           deleteWatchers.update(watcherKey, sender())
 
     case request: UnwatchEndpoint =>
+      logging.info(this, s"unwatch endpoint: $request")
       val watcherKey = WatcherKey(request.watchKey, request.watchName)
       if (request.isPrefix) {
         prefixPutWatchers.remove(watcherKey)

core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/FPCPoolBalancer.scala

@@ -174,7 +174,7 @@ class FPCPoolBalancer(config: WhiskConfig,
     if (retryCount >= 0)
       scheduler
         .getRemoteRef(QueueManager.actorName)
-        .ask(CreateQueue(invocationNamespace, fullyQualifiedEntityName.copy(binding = None), revision, actionMetaData))
+        .ask(CreateQueue(invocationNamespace, fullyQualifiedEntityName, revision, actionMetaData))
         .mapTo[CreateQueueResponse]
         .onComplete {
           case Success(_) =>

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/FunctionPullingContainerPool.scala

@@ -680,7 +680,13 @@ class FunctionPullingContainerPool(
       case Some(((proxy, data), containerState)) =>
         // record creationMessage so when container created failed, we can send failed message to scheduler
         creationMessages.getOrElseUpdate(proxy, create)
-        proxy ! Initialize(create.invocationNamespace, executable, create.schedulerHost, create.rpcPort, create.transid)
+        proxy ! Initialize(
+          create.invocationNamespace,
+          create.action,
+          executable,
+          create.schedulerHost,
+          create.rpcPort,
+          create.transid)
         inProgressPool = inProgressPool + (proxy -> data)
         logContainerStart(create, executable.toWhiskAction, containerState)
 

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/FunctionPullingContainerProxy.scala

@@ -68,6 +68,7 @@ case class InitCodeCompleted(data: WarmData)
 
 // Events received by the actor
 case class Initialize(invocationNamespace: String,
+                      fqn: FullyQualifiedEntityName,
                       action: ExecutableWhiskAction,
                       schedulerHost: String,
                       rpcPort: Int,
@@ -251,7 +252,7 @@ class FunctionPullingContainerProxy(
             clientProxyFactory(
               context,
               job.invocationNamespace,
-              job.action.fullyQualifiedName(true),
+              job.fqn, // include binding field
               job.action.rev,
               job.schedulerHost,
               job.rpcPort,
@@ -292,7 +293,7 @@ class FunctionPullingContainerProxy(
         clientProxyFactory(
           context,
           job.invocationNamespace,
-          job.action.fullyQualifiedName(true),
+          job.fqn, // include binding field
           job.action.rev,
           job.schedulerHost,
           job.rpcPort,

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/InvokerHealthManager.scala

@@ -172,7 +172,8 @@ class InvokerHealthManager(instanceId: InvokerInstanceId,
 
     WhiskAction.get(entityStore, docId).onComplete {
       case Success(action) =>
-        val initialize = Initialize(namespace, action.toExecutableWhiskAction.get, "", 0, transid)
+        val initialize =
+          Initialize(namespace, action.fullyQualifiedName(true), action.toExecutableWhiskAction.get, "", 0, transid)
         startHealthAction(initialize, manager)
       case Failure(t) => logging.error(this, s"get health action error: ${t.getMessage}")
     }

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/QueueManager.scala

@@ -179,7 +179,8 @@ class QueueManager(
       }
 
     case RecoverQueue(msg, action, actionMetaData) =>
-      QueuePool.keys.find(_.docInfo.id == action.toDocId) match {
+      QueuePool.keys.find(k =>
+        k.invocationNamespace == msg.user.namespace.name.asString && k.docInfo.id == action.toDocId) match {
         // queue is already recovered or a newer queue is created, send msg to new queue
         case Some(key) if key.docInfo.rev >= msg.revision =>
           QueuePool.get(key) match {
@@ -358,10 +359,12 @@ class QueueManager(
   }
 
   private def handleCycle(msg: ActivationMessage)(implicit transid: TransactionId): Future[Any] = {
-    logging.warn(this, s"queue for ${msg.action} doesn't exist in memory but exist in etcd, recovering...")
+    logging.warn(
+      this,
+      s"queue for ${msg.user.namespace.name.asString}/${msg.action} doesn't exist in memory but exist in etcd, recovering...")
     val start = transid.started(this, LoggingMarkers.SCHEDULER_QUEUE_RECOVER)
 
-    logging.info(this, s"Recover a queue for ${msg.action},")
+    logging.info(this, s"Recover a queue for ${msg.user.namespace.name.asString}/${msg.action},")
     getWhiskActionMetaData(entityStore, msg.action.toDocId, msg.revision, false)
       .map { actionMetaData: WhiskActionMetaData =>
         actionMetaData.toExecutableWhiskAction match {
@@ -397,7 +400,7 @@ class QueueManager(
       logging.warn(
         this,
         s"[${msg.activationId}] the activation message has not been scheduled for ${queueManagerConfig.maxSchedulingTime.toSeconds} sec")
-      completeErrorActivation(msg, "The activation has not been processed")
+      completeErrorActivation(msg, "The activation has not been processed: too old activation is arrived.")
     } else {
       QueuePool.get(MemoryQueueKey(msg.user.namespace.name.asString, msg.action.toDocId.asDocInfo(msg.revision))) match {
         case Some(memoryQueueValue) if memoryQueueValue.isLeader =>
@@ -466,13 +469,15 @@ class QueueManager(
         .recoverWith {
           case t =>
             logging.warn(this, s"[${msg.activationId}] failed to parse endpoints (${t.getMessage})")
-            completeErrorActivation(msg, "The activation has not been processed")
+            completeErrorActivation(
+              msg,
+              "The activation has not been processed: failed to parse the scheduler endpoint.")
         }
 
     } recoverWith {
       case t =>
         logging.warn(this, s"[${msg.activationId}] activation has been dropped (${t.getMessage})")
-        completeErrorActivation(msg, "The activation has not been processed")
+        completeErrorActivation(msg, "The activation has not been processed: failed to get the queue endpoint.")
     }
   }
 

tests/src/test/scala/org/apache/openwhisk/core/containerpool/v2/test/FunctionPullingContainerPoolTests.scala

@@ -227,12 +227,12 @@ class FunctionPullingContainerPoolTests
 
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(1).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -254,7 +254,7 @@ class FunctionPullingContainerPoolTests
     // Start first action
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction) // 1 * stdMemory taken
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // Send second action to the pool
@@ -267,7 +267,7 @@ class FunctionPullingContainerPoolTests
     pool ! CreationContainer(creationMessageLarge.copy(revision = bigDoc.rev), bigWhiskAction)
     // Second container should run now
     containers(1).expectMsgPF() {
-      case Initialize(invocationNamespace, bigExecuteAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, bigExecuteAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -307,7 +307,7 @@ class FunctionPullingContainerPoolTests
     pool ! Enable
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction) // 1 * stdMemory taken
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -328,7 +328,7 @@ class FunctionPullingContainerPoolTests
     (0 to 10).foreach(_ => pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)) // 11 * stdMemory taken)
     (0 to 10).foreach(i => {
       containers(i).expectMsgPF() {
-        case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+        case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
       }
       // create 5 container in busy pool, and 6 in warmed pool
       if (i < 5)
@@ -526,7 +526,7 @@ class FunctionPullingContainerPoolTests
     // the prewarm container with matched memory should be chose
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // prewarm a new container
@@ -537,7 +537,7 @@ class FunctionPullingContainerPoolTests
     // the prewarm container with bigger memory should not be chose
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(3).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -566,7 +566,7 @@ class FunctionPullingContainerPoolTests
     // the prewarm container with smallest memory should be chose
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // prewarm a new container
@@ -577,7 +577,7 @@ class FunctionPullingContainerPoolTests
     // the prewarm container with bigger memory should be chose
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(1).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // prewarm a new container
@@ -589,7 +589,7 @@ class FunctionPullingContainerPoolTests
     // a new container should be created
     pool ! CreationContainer(creationMessageLarge.copy(revision = doc.rev), bigWhiskAction)
     containers(4).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // no new prewarmed container should be created
@@ -616,7 +616,7 @@ class FunctionPullingContainerPoolTests
 
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(1).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -640,7 +640,7 @@ class FunctionPullingContainerPoolTests
 
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(1).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -674,24 +674,24 @@ class FunctionPullingContainerPoolTests
     // the revision doesn't match, create 1 container
     pool ! CreationContainer(creationMessage, whiskAction)
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // the invocation namespace doesn't match, create 1 container
     pool ! CreationContainer(creationMessage.copy(invocationNamespace = "otherNamespace"), whiskAction)
     containers(1).expectMsgPF() {
-      case Initialize("otherNamespace", executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize("otherNamespace", fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     container.expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // warmed container is occupied, create 1 more container
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(2).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -726,7 +726,7 @@ class FunctionPullingContainerPoolTests
     // choose the warmed container
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     container.expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // warmed container is failed to resume
@@ -743,7 +743,7 @@ class FunctionPullingContainerPoolTests
 
     // then a new container will be created
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -806,7 +806,7 @@ class FunctionPullingContainerPoolTests
 
     // a new container will be created
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
   }
 
@@ -850,7 +850,7 @@ class FunctionPullingContainerPoolTests
 
     pool ! CreationContainer(actualCreationMessage, whiskAction)
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     containers(0).send(pool, Initialized(initializedData)) // container is initialized
 
@@ -898,7 +898,7 @@ class FunctionPullingContainerPoolTests
     // choose the warmed container
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     container.expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     pool.tell(
       Resumed(
@@ -956,7 +956,7 @@ class FunctionPullingContainerPoolTests
 
     pool ! CreationContainer(actualCreationMessage, whiskAction)
     containers(0).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     containers(0).send(pool, ContainerRemoved(true)) // the container0 init failed or create container failed
 
@@ -1127,11 +1127,11 @@ class FunctionPullingContainerPoolTests
     // 2 cold start happened
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(2).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(3).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     // Make sure AdjustPrewarmedContainer is sent by ContainerPool's scheduler after prewarmExpirationCheckIntervel time
     Thread.sleep(prewarmExpirationCheckIntervel.toMillis)
@@ -1165,23 +1165,23 @@ class FunctionPullingContainerPoolTests
     // 5 code start happened(5 > maxCount)
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(6).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(7).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(8).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(9).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
     pool ! CreationContainer(creationMessage.copy(revision = doc.rev), whiskAction)
     containers(10).expectMsgPF() {
-      case Initialize(invocationNamespace, executeAction, schedulerHost, rpcPort, _) => true
+      case Initialize(invocationNamespace, fqn, executeAction, schedulerHost, rpcPort, _) => true
     }
 
     // Make sure AdjustPrewarmedContainer is sent by ContainerPool's scheduler after prewarmExpirationCheckIntervel time