Implement action versioning

Author: jiangpengcheng

ansible/files/whisks_design_document_for_entities_db_v2.1.0.json

@@ -21,6 +21,9 @@
     "triggers": {
       "map": "function (doc) {\n  var PATHSEP = \"/\";\n  var isTrigger = function (doc) { return (doc.exec === undefined && doc.binding === undefined && doc.parameters !== undefined) };\n  if (isTrigger(doc)) try {\n    var ns = doc.namespace.split(PATHSEP);\n    var root = ns[0];\n    var value = {\n      namespace: doc.namespace,\n      name: doc.name,\n      version: doc.version,\n      publish: doc.publish,\n      annotations: doc.annotations,\n      updated: doc.updated\n    };\n    emit([doc.namespace, doc.updated], value);\n    if (root !== doc.namespace) {\n      emit([root, doc.updated], value);\n    }\n  } catch (e) {}\n}",
       "reduce": "_count"
+    },
+    "action-versions": {
+      "map": "function (doc) {\n  var isAction = function (doc) { return (doc.exec !== undefined) };\n  if (isAction(doc)) try {\n    var value = {\n      namespace: doc.namespace,\n      name: doc.name,\n      id: doc._id,\n      version: doc.version,\n    };\n    emit([doc.namespace + \"/\" + doc.name], value);\n  } catch (e) {}\n}"
     }
   }
-}
+}

common/scala/src/main/scala/org/apache/openwhisk/core/connector/Message.scala

@@ -53,6 +53,7 @@ case class ActivationMessage(override val transid: TransactionId,
                              revision: DocRevision,
                              user: Identity,
                              activationId: ActivationId,
+                             actionId: DocId,
                              rootControllerIndex: ControllerInstanceId,
                              blocking: Boolean,
                              content: Option[JsObject],
@@ -171,7 +172,7 @@ object ActivationMessage extends DefaultJsonProtocol {
   def parse(msg: String) = Try(serdes.read(msg.parseJson))
 
   private implicit val fqnSerdes = FullyQualifiedEntityName.serdes
-  implicit val serdes = jsonFormat11(ActivationMessage.apply)
+  implicit val serdes = jsonFormat12(ActivationMessage.apply)
 }
 
 object CombinedCompletionAndResultMessage extends DefaultJsonProtocol {

common/scala/src/main/scala/org/apache/openwhisk/core/database/RemoteCacheInvalidation.scala

@@ -33,13 +33,16 @@ import org.apache.openwhisk.core.WhiskConfig
 import org.apache.openwhisk.core.connector.Message
 import org.apache.openwhisk.core.connector.MessageFeed
 import org.apache.openwhisk.core.connector.MessagingProvider
-import org.apache.openwhisk.core.entity.CacheKey
-import org.apache.openwhisk.core.entity.ControllerInstanceId
-import org.apache.openwhisk.core.entity.WhiskAction
-import org.apache.openwhisk.core.entity.WhiskActionMetaData
-import org.apache.openwhisk.core.entity.WhiskPackage
-import org.apache.openwhisk.core.entity.WhiskRule
-import org.apache.openwhisk.core.entity.WhiskTrigger
+import org.apache.openwhisk.core.entity.{
+  CacheKey,
+  ControllerInstanceId,
+  WhiskAction,
+  WhiskActionMetaData,
+  WhiskActionVersionList,
+  WhiskPackage,
+  WhiskRule,
+  WhiskTrigger
+}
 import org.apache.openwhisk.spi.SpiLoader
 
 case class CacheInvalidationMessage(key: CacheKey, instanceId: String) extends Message {
@@ -92,6 +95,7 @@ class RemoteCacheInvalidation(config: WhiskConfig, component: String, instance:
           WhiskPackage.removeId(msg.key)
           WhiskRule.removeId(msg.key)
           WhiskTrigger.removeId(msg.key)
+          WhiskActionVersionList.removeId(msg.key)
         }
       }
       case Failure(t) => logging.error(this, s"failed processing message: $raw with $t")

common/scala/src/main/scala/org/apache/openwhisk/core/entity/SemVer.scala

@@ -78,7 +78,7 @@ protected[core] object SemVer {
    * @return SemVer instance
    * @thrown IllegalArgumentException if string is not a valid semantic version
    */
-  protected[entity] def apply(str: String): SemVer = {
+  protected[core] def apply(str: String): SemVer = {
     try {
       val parts = if (str != null && str.nonEmpty) str.split('.') else Array[String]()
       val major = if (parts.size >= 1) parts(0).toInt else 0

common/scala/src/main/scala/org/apache/openwhisk/core/entity/WhiskAction.scala

@@ -29,10 +29,14 @@ import scala.concurrent.Future
 import scala.util.{Failure, Success, Try}
 import spray.json._
 import spray.json.DefaultJsonProtocol._
-import org.apache.openwhisk.common.TransactionId
-import org.apache.openwhisk.core.database.ArtifactStore
-import org.apache.openwhisk.core.database.DocumentFactory
-import org.apache.openwhisk.core.database.CacheChangeNotification
+import org.apache.openwhisk.common.{Logging, TransactionId}
+import org.apache.openwhisk.core.database.{
+  ArtifactStore,
+  CacheChangeNotification,
+  DocumentFactory,
+  MultipleReadersSingleWriterCache,
+  StaleParameter
+}
 import org.apache.openwhisk.core.entity.Attachments._
 import org.apache.openwhisk.core.entity.types.EntityStore
 
@@ -106,8 +110,10 @@ abstract class WhiskActionLike(override val name: EntityName) extends WhiskEntit
       "annotations" -> annotations.toJson)
 }
 
-abstract class WhiskActionLikeMetaData(override val name: EntityName) extends WhiskActionLike(name) {
+abstract class WhiskActionLikeMetaData(override val name: EntityName, val docId: DocId) extends WhiskActionLike(name) {
   override def exec: ExecMetaDataBase
+
+  override def docid = docId
 }
 
 /**
@@ -143,6 +149,8 @@ case class WhiskAction(namespace: EntityPath,
   require(exec != null, "exec undefined")
   require(limits != null, "limits undefined")
 
+  override def docid = DocId(fullyQualifiedName(true).asString)
+
   /**
    * Merges parameters (usually from package) with existing action parameters.
    * Existing parameters supersede those in p.
@@ -173,7 +181,7 @@ case class WhiskAction(namespace: EntityPath,
   def toExecutableWhiskAction: Option[ExecutableWhiskAction] = exec match {
     case codeExec: CodeExec[_] =>
       Some(
-        ExecutableWhiskAction(namespace, name, codeExec, parameters, limits, version, publish, annotations)
+        ExecutableWhiskAction(namespace, name, docid, codeExec, parameters, limits, version, publish, annotations)
           .revision[ExecutableWhiskAction](rev))
     case _ => None
   }
@@ -198,6 +206,7 @@ case class WhiskAction(namespace: EntityPath,
 @throws[IllegalArgumentException]
 case class WhiskActionMetaData(namespace: EntityPath,
                                override val name: EntityName,
+                               override val docId: DocId,
                                exec: ExecMetaDataBase,
                                parameters: Parameters = Parameters(),
                                limits: ActionLimits = ActionLimits(),
@@ -206,7 +215,7 @@ case class WhiskActionMetaData(namespace: EntityPath,
                                annotations: Parameters = Parameters(),
                                override val updated: Instant = WhiskEntity.currentMillis(),
                                binding: Option[EntityPath] = None)
-    extends WhiskActionLikeMetaData(name) {
+    extends WhiskActionLikeMetaData(name, docId) {
 
   require(exec != null, "exec undefined")
   require(limits != null, "limits undefined")
@@ -238,6 +247,7 @@ case class WhiskActionMetaData(namespace: EntityPath,
         ExecutableWhiskActionMetaData(
           namespace,
           name,
+          docId,
           execMetaData,
           parameters,
           limits,
@@ -276,6 +286,7 @@ case class WhiskActionMetaData(namespace: EntityPath,
 @throws[IllegalArgumentException]
 case class ExecutableWhiskAction(namespace: EntityPath,
                                  override val name: EntityName,
+                                 docId: DocId,
                                  exec: CodeExec[_],
                                  parameters: Parameters = Parameters(),
                                  limits: ActionLimits = ActionLimits(),
@@ -322,20 +333,21 @@ case class ExecutableWhiskAction(namespace: EntityPath,
 @throws[IllegalArgumentException]
 case class ExecutableWhiskActionMetaData(namespace: EntityPath,
                                          override val name: EntityName,
+                                         override val docId: DocId,
                                          exec: ExecMetaData,
                                          parameters: Parameters = Parameters(),
                                          limits: ActionLimits = ActionLimits(),
                                          version: SemVer = SemVer(),
                                          publish: Boolean = false,
                                          annotations: Parameters = Parameters(),
                                          binding: Option[EntityPath] = None)
-    extends WhiskActionLikeMetaData(name) {
+    extends WhiskActionLikeMetaData(name, docId) {
 
   require(exec != null, "exec undefined")
   require(limits != null, "limits undefined")
 
   def toWhiskAction =
-    WhiskActionMetaData(namespace, name, exec, parameters, limits, version, publish, annotations, updated)
+    WhiskActionMetaData(namespace, name, docId, exec, parameters, limits, version, publish, annotations, updated)
       .revision[WhiskActionMetaData](rev)
 
   /**
@@ -346,6 +358,74 @@ case class ExecutableWhiskActionMetaData(namespace: EntityPath,
 
 }
 
+case class WhiskActionVersion(id: String, namespace: EntityPath, name: EntityName, version: SemVer)
+
+object WhiskActionVersion {
+  val serdes = jsonFormat4(WhiskActionVersion.apply)
+}
+
+case class WhiskActionVersionList(namespace: EntityPath, name: EntityName, versions: Map[SemVer, String]) {
+  def matchedDocId(version: Option[SemVer]): Option[DocId] = {
+    version match {
+      case Some(ver) =>
+        versions.get(ver).map(DocId(_))
+      case None if versions.nonEmpty =>
+        Some(DocId(versions.maxBy(_._1.toString)._2))
+      case _ =>
+        None
+    }
+  }
+}
+
+object WhiskActionVersionList extends MultipleReadersSingleWriterCache[WhiskActionVersionList, DocInfo] {
+  lazy val viewName = WhiskQueries.entitiesView(collection = "action-versions").name
+
+  def cacheKey(action: FullyQualifiedEntityName): CacheKey = {
+    CacheKey(action.fullPath.asString)
+  }
+
+  def get(action: FullyQualifiedEntityName, datastore: EntityStore)(
+    implicit transId: TransactionId): Future[WhiskActionVersionList] = {
+    implicit val logger: Logging = datastore.logging
+    implicit val ec = datastore.executionContext
+
+    val key = List(action.fullPath.asString)
+    cacheLookup(
+      cacheKey(action),
+      datastore
+        .query(
+          viewName,
+          startKey = key,
+          endKey = key,
+          skip = 0,
+          limit = 0,
+          includeDocs = false,
+          descending = false,
+          reduce = false,
+          stale = StaleParameter.No)
+        .map { result =>
+          val values = result.map { row =>
+            row.fields("value").asJsObject()
+          }
+          val mappings = values
+            .map(WhiskActionVersion.serdes.read(_))
+            .map { actionVersion =>
+              (actionVersion.version, actionVersion.id)
+            }
+            .toMap
+          WhiskActionVersionList(action.namespace.toPath, action.name, mappings)
+        })
+  }
+
+  // delete cache
+  def deleteCache(action: FullyQualifiedEntityName)(implicit transId: TransactionId,
+                                                    ec: ExecutionContext,
+                                                    logger: Logging,
+                                                    notifier: Option[CacheChangeNotification]) = {
+    cacheInvalidate(cacheKey(action), Future.successful(()))
+  }
+}
+
 object WhiskAction extends DocumentFactory[WhiskAction] with WhiskEntityQueries[WhiskAction] with DefaultJsonProtocol {
   import WhiskActivation.instantSerdes
   val execFieldName = "exec"
@@ -410,7 +490,11 @@ object WhiskAction extends DocumentFactory[WhiskAction] with WhiskEntityQueries[
             old)
       }
     } match {
-      case Success(f) => f
+      case Success(f) =>
+        implicit val ec = db.executionContext
+        implicit val logger = db.logging
+        WhiskActionVersionList.deleteCache(doc.fullyQualifiedName(false))
+        f
       case Failure(f) => Future.failed(f)
     }
   }
@@ -532,14 +616,18 @@ object WhiskAction extends DocumentFactory[WhiskAction] with WhiskEntityQueries[
    * If it's the actual package, use its name directly as the package path name.
    * While traversing the package bindings, merge the parameters.
    */
-  def resolveActionAndMergeParameters(entityStore: EntityStore, fullyQualifiedName: FullyQualifiedEntityName)(
-    implicit ec: ExecutionContext,
-    transid: TransactionId): Future[WhiskAction] = {
+  def resolveActionAndMergeParameters(
+    entityStore: EntityStore,
+    fullyQualifiedName: FullyQualifiedEntityName,
+    version: Option[SemVer] = None)(implicit ec: ExecutionContext, transid: TransactionId): Future[WhiskAction] = {
     // first check that there is a package to be resolved
     val entityPath = fullyQualifiedName.path
     if (entityPath.defaultPackage) {
       // this is the default package, nothing to resolve
-      WhiskAction.get(entityStore, fullyQualifiedName.toDocId)
+      WhiskActionVersionList.get(fullyQualifiedName, entityStore).flatMap { result =>
+        val docId = result.matchedDocId(version).getOrElse(fullyQualifiedName.toDocId)
+        WhiskAction.get(entityStore, docId)
+      }
     } else {
       // there is a package to be resolved
       val pkgDocid = fullyQualifiedName.path.toDocId
@@ -548,8 +636,12 @@ object WhiskAction extends DocumentFactory[WhiskAction] with WhiskEntityQueries[
       wp flatMap { resolvedPkg =>
         // fully resolved name for the action
         val fqnAction = resolvedPkg.fullyQualifiedName(withVersion = false).add(actionName)
+        val action = WhiskActionVersionList.get(fqnAction, entityStore).flatMap { result =>
+          val docId = result.matchedDocId(version).getOrElse(fqnAction.toDocId)
+          WhiskAction.get(entityStore, docId)
+        }
         // get the whisk action associate with it and inherit the parameters from the package/binding
-        WhiskAction.get(entityStore, fqnAction.toDocId) map {
+        action map {
           _.inherit(resolvedPkg.parameters)
         }
       }
@@ -571,6 +663,7 @@ object WhiskActionMetaData
     WhiskActionMetaData.apply,
     "namespace",
     "name",
+    "_id",
     "exec",
     "parameters",
     "limits",
@@ -611,14 +704,19 @@ object WhiskActionMetaData
    * If it's the actual package, use its name directly as the package path name.
    * While traversing the package bindings, merge the parameters.
    */
-  def resolveActionAndMergeParameters(entityStore: EntityStore, fullyQualifiedName: FullyQualifiedEntityName)(
+  def resolveActionAndMergeParameters(entityStore: EntityStore,
+                                      fullyQualifiedName: FullyQualifiedEntityName,
+                                      version: Option[SemVer] = None)(
     implicit ec: ExecutionContext,
     transid: TransactionId): Future[WhiskActionMetaData] = {
     // first check that there is a package to be resolved
     val entityPath = fullyQualifiedName.path
     if (entityPath.defaultPackage) {
       // this is the default package, nothing to resolve
-      WhiskActionMetaData.get(entityStore, fullyQualifiedName.toDocId)
+      WhiskActionVersionList.get(fullyQualifiedName, entityStore).flatMap { result =>
+        val docId = result.matchedDocId(version).getOrElse(fullyQualifiedName.toDocId)
+        WhiskActionMetaData.get(entityStore, docId)
+      }
     } else {
       // there is a package to be resolved
       val pkgDocid = fullyQualifiedName.path.toDocId
@@ -627,8 +725,12 @@ object WhiskActionMetaData
       wp flatMap { resolvedPkg =>
         // fully resolved name for the action
         val fqnAction = resolvedPkg.fullyQualifiedName(withVersion = false).add(actionName)
+        val action = WhiskActionVersionList.get(fqnAction, entityStore).flatMap { result =>
+          val docId = result.matchedDocId(version).getOrElse(fqnAction.toDocId)
+          WhiskActionMetaData.get(entityStore, docId)
+        }
         // get the whisk action associate with it and inherit the parameters from the package/binding
-        WhiskActionMetaData.get(entityStore, fqnAction.toDocId) map {
+        action map {
           _.inherit(
             resolvedPkg.parameters,
             if (fullyQualifiedName.path.equals(resolvedPkg.fullPath)) None

common/scala/src/main/scala/org/apache/openwhisk/core/entity/WhiskEntity.scala

@@ -58,7 +58,7 @@ abstract class WhiskEntity protected[entity] (en: EntityName, val entityType: St
     FullyQualifiedEntityName(namespace, en, if (withVersion) Some(version) else None)
 
   /** The primary key for the entity in the datastore */
-  override final def docid = fullyQualifiedName(false).toDocId
+  override def docid = fullyQualifiedName(false).toDocId
 
   /**
    * Returns a JSON object with the fields specific to this abstract class.