Consider binding action when creating or recovering queue.

Author: style95

common/scala/src/main/scala/org/apache/openwhisk/common/Logging.scala

@@ -593,7 +593,8 @@ object LoggingMarkers {
   val SCHEDULER_KAFKA = LogMarkerToken(scheduler, kafka, start)(MeasurementUnit.time.milliseconds)
   val SCHEDULER_KAFKA_WAIT_TIME =
     LogMarkerToken(scheduler, "kafkaWaitTime", counter)(MeasurementUnit.none)
-  def SCHEDULER_WAIT_TIME(action: String) = LogMarkerToken(scheduler, "waitTime", counter, Some(action), Map("action" -> action))(MeasurementUnit.none)
+  def SCHEDULER_WAIT_TIME(action: String) =
+    LogMarkerToken(scheduler, "waitTime", counter, Some(action), Map("action" -> action))(MeasurementUnit.none)
 
   def SCHEDULER_KEEP_ALIVE(leaseId: Long) =
     LogMarkerToken(scheduler, "keepAlive", counter, None, Map("leaseId" -> leaseId.toString))(MeasurementUnit.none)

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/MemoryQueue.scala

@@ -35,7 +35,12 @@ import org.apache.openwhisk.core.etcd.EtcdClient
 import org.apache.openwhisk.core.etcd.EtcdKV.ContainerKeys.containerPrefix
 import org.apache.openwhisk.core.etcd.EtcdKV.{ContainerKeys, QueueKeys, ThrottlingKeys}
 import org.apache.openwhisk.core.scheduler.grpc.{GetActivation, ActivationResponse => GetActivationResponse}
-import org.apache.openwhisk.core.scheduler.message.{ContainerCreation, ContainerDeletion, FailedCreationJob, SuccessfulCreationJob}
+import org.apache.openwhisk.core.scheduler.message.{
+  ContainerCreation,
+  ContainerDeletion,
+  FailedCreationJob,
+  SuccessfulCreationJob
+}
 import org.apache.openwhisk.core.scheduler.{SchedulerEndpoints, SchedulingConfig}
 import org.apache.openwhisk.core.service._
 import org.apache.openwhisk.http.Messages.{namespaceLimitUnderZero, tooManyConcurrentRequests}
@@ -47,7 +52,7 @@ import scala.annotation.tailrec
 import scala.collection.immutable.Queue
 import scala.collection.mutable
 import scala.concurrent.duration._
-import scala.concurrent.{ExecutionContextExecutor, Future, Promise, duration}
+import scala.concurrent.{duration, ExecutionContextExecutor, Future, Promise}
 import scala.util.{Failure, Success}
 
 // States
@@ -766,7 +771,9 @@ class MemoryQueue(private val etcdClient: EtcdClient,
       activation.transid)
 
     val totalTimeInScheduler = Interval(activation.transid.meta.start, Instant.now()).duration
-    MetricEmitter.emitHistogramMetric(LoggingMarkers.SCHEDULER_WAIT_TIME(action.asString), totalTimeInScheduler.toMillis)
+    MetricEmitter.emitHistogramMetric(
+      LoggingMarkers.SCHEDULER_WAIT_TIME(action.asString),
+      totalTimeInScheduler.toMillis)
 
     val activationResponse =
       if (isWhiskError)
@@ -938,7 +945,9 @@ class MemoryQueue(private val etcdClient: EtcdClient,
     takeUncompletedRequest()
       .map { res =>
         val totalTimeInScheduler = Interval(msg.transid.meta.start, Instant.now()).duration
-        MetricEmitter.emitHistogramMetric(LoggingMarkers.SCHEDULER_WAIT_TIME(action.asString), totalTimeInScheduler.toMillis)
+        MetricEmitter.emitHistogramMetric(
+          LoggingMarkers.SCHEDULER_WAIT_TIME(action.asString),
+          totalTimeInScheduler.toMillis)
         res.trySuccess(Right(msg))
         in.decrementAndGet()
         stay
@@ -960,7 +969,9 @@ class MemoryQueue(private val etcdClient: EtcdClient,
         this,
         s"[$invocationNamespace:$action:$stateName] Get activation request ${request.containerId}, send one message: ${msg.activationId}")
       val totalTimeInScheduler = Interval(msg.transid.meta.start, Instant.now()).duration
-      MetricEmitter.emitHistogramMetric(LoggingMarkers.SCHEDULER_WAIT_TIME(action.asString), totalTimeInScheduler.toMillis)
+      MetricEmitter.emitHistogramMetric(
+        LoggingMarkers.SCHEDULER_WAIT_TIME(action.asString),
+        totalTimeInScheduler.toMillis)
 
       sender ! GetActivationResponse(Right(msg))
       tryDisableActionThrottling()

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.")
     }
   }