Skip to content

Commit a75978c

Browse files
jiangpengchengjiangpengcheng
committed
Some updates:
1. provide a migrate script to copy old actions to new docs with new id 2. use a separate entity to save the default version for action
1 parent 5c69dbf commit a75978c

File tree

11 files changed

+453
-169
lines changed

11 files changed

+453
-169
lines changed

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

Lines changed: 48 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -34,8 +34,10 @@ import org.apache.openwhisk.core.database.{
3434
ArtifactStore,
3535
CacheChangeNotification,
3636
DocumentFactory,
37+
EvictionPolicy,
3738
MultipleReadersSingleWriterCache,
38-
StaleParameter
39+
StaleParameter,
40+
WriteTime
3941
}
4042
import org.apache.openwhisk.core.entity.Attachments._
4143
import org.apache.openwhisk.core.entity.types.EntityStore
@@ -358,22 +360,22 @@ case class ExecutableWhiskActionMetaData(namespace: EntityPath,
358360

359361
}
360362

361-
case class WhiskActionVersion(id: String, namespace: EntityPath, name: EntityName, version: SemVer, publish: Boolean)
363+
case class WhiskActionVersion(id: String, namespace: EntityPath, name: EntityName, version: SemVer)
362364

363365
object WhiskActionVersion {
364-
val serdes = jsonFormat(WhiskActionVersion.apply, "id", "namespace", "name", "version", "publish")
366+
val serdes = jsonFormat(WhiskActionVersion.apply, "id", "namespace", "name", "version")
365367
}
366368

367369
case class WhiskActionVersionList(namespace: EntityPath,
368370
name: EntityName,
369371
versions: Map[SemVer, String],
370-
defaultVersion: Option[String]) {
372+
defaultVersion: Option[SemVer]) {
371373
def matchedDocId(version: Option[SemVer]): Option[DocId] = {
372374
version match {
373375
case Some(ver) =>
374-
versions.get(ver).map(DocId(_))
376+
Some(DocId(s"$namespace/$name@$ver"))
375377
case None if defaultVersion.nonEmpty =>
376-
versions.get(SemVer(defaultVersion.get)).map(DocId(_))
378+
Some(DocId(s"$namespace/$name@${defaultVersion.get}"))
377379
case None if versions.nonEmpty =>
378380
Some(DocId(versions.maxBy(_._1)._2))
379381
case _ =>
@@ -383,8 +385,10 @@ case class WhiskActionVersionList(namespace: EntityPath,
383385
}
384386

385387
object WhiskActionVersionList extends MultipleReadersSingleWriterCache[WhiskActionVersionList, DocInfo] {
388+
override val evictionPolicy: EvictionPolicy = WriteTime
386389
val collectionName = "action-versions"
387390
lazy val viewName = WhiskQueries.entitiesView(collection = collectionName).name
391+
implicit val serdes = jsonFormat(WhiskActionVersionList.apply, "namespace", "name", "versions", "defaultVersion")
388392

389393
def cacheKey(action: FullyQualifiedEntityName): CacheKey = {
390394
CacheKey(action.fullPath.asString)
@@ -421,38 +425,30 @@ object WhiskActionVersionList extends MultipleReadersSingleWriterCache[WhiskActi
421425
}
422426
.toMap
423427
val defaultVersion = if (result.nonEmpty) {
424-
val doc = result.head.fields.getOrElse("doc", JsNull)
425-
if (doc != JsNull) doc.asJsObject.fields.get("default").map(_.convertTo[String])
426-
else
427-
None
428+
result.head.fields.get("doc") match {
429+
case Some(value) => Try { value.asJsObject.fields.get("default").map(_.convertTo[SemVer]) } getOrElse None
430+
case None => None
431+
}
428432
} else None
429433
WhiskActionVersionList(action.namespace.toPath, action.name, mappings, defaultVersion)
430434
},
431435
fromCache)
432436
}
433437

434-
def getMatchedDocId(
435-
action: FullyQualifiedEntityName,
436-
version: Option[SemVer],
437-
datastore: EntityStore,
438-
tryAgain: Boolean = true)(implicit transId: TransactionId, ec: ExecutionContext): Future[Option[DocId]] = {
439-
get(action, datastore).flatMap { res =>
440-
val docId = version match {
441-
case Some(ver) =>
442-
res.versions.get(ver).map(DocId(_))
438+
def getMatchedDocId(action: FullyQualifiedEntityName, version: Option[SemVer], datastore: EntityStore)(
439+
implicit transId: TransactionId,
440+
ec: ExecutionContext): Future[Option[DocId]] = {
441+
get(action, datastore).map { res =>
442+
version match {
443+
case Some(_) =>
444+
Some(DocId(action.copy(version = version).asString))
443445
case None if res.defaultVersion.nonEmpty =>
444-
res.versions.get(SemVer(res.defaultVersion.get)).map(DocId(_))
446+
Some(DocId(action.copy(version = res.defaultVersion).asString))
445447
case None if res.versions.nonEmpty =>
446448
Some(DocId(res.versions.maxBy(_._1)._2))
447449
case _ =>
448450
None
449451
}
450-
// there may be a chance that database is updated while cache is not, we need to invalidate cache and try again
451-
if (docId.isEmpty && tryAgain) {
452-
WhiskActionVersionList.removeId(cacheKey(action))
453-
getMatchedDocId(action, version, datastore, false)
454-
} else
455-
Future.successful(docId)
456452
}
457453
}
458454

@@ -465,6 +461,32 @@ object WhiskActionVersionList extends MultipleReadersSingleWriterCache[WhiskActi
465461
}
466462
}
467463

464+
object WhiskActionDefaultVersion extends DocumentFactory[WhiskActionDefaultVersion] {
465+
import WhiskActivation.instantSerdes
466+
implicit val serdes = jsonFormat(WhiskActionDefaultVersion.apply, "namespace", "name", "default", "updated")
467+
}
468+
469+
case class WhiskActionDefaultVersion(namespace: EntityPath,
470+
override val name: EntityName,
471+
default: Option[SemVer] = None,
472+
override val updated: Instant = WhiskEntity.currentMillis())
473+
extends WhiskEntity(name, "action-default-version") {
474+
475+
/**
476+
* The representation as JSON, e.g. for REST calls. Does not include id/rev.
477+
*/
478+
override def toJson: JsObject = WhiskActionDefaultVersion.serdes.write(this).asJsObject
479+
480+
/**
481+
* Gets unique document identifier for the document.
482+
*/
483+
override def docid: DocId = new DocId(namespace + EntityPath.PATHSEP + name + "/default")
484+
485+
override val version: SemVer = SemVer()
486+
override val publish: Boolean = true
487+
override val annotations: Parameters = Parameters()
488+
}
489+
468490
object WhiskAction extends DocumentFactory[WhiskAction] with WhiskEntityQueries[WhiskAction] with DefaultJsonProtocol {
469491
import WhiskActivation.instantSerdes
470492

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

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -125,13 +125,14 @@ object WhiskEntity {
125125
object WhiskDocumentReader extends DocumentReader {
126126
override def read[A](ma: Manifest[A], value: JsValue) = {
127127
val doc = ma.runtimeClass match {
128-
case x if x == classOf[WhiskAction] => WhiskAction.serdes.read(value)
129-
case x if x == classOf[WhiskActionMetaData] => WhiskActionMetaData.serdes.read(value)
130-
case x if x == classOf[WhiskPackage] => WhiskPackage.serdes.read(value)
131-
case x if x == classOf[WhiskActivation] => WhiskActivation.serdes.read(value)
132-
case x if x == classOf[WhiskTrigger] => WhiskTrigger.serdes.read(value)
133-
case x if x == classOf[WhiskRule] => WhiskRule.serdes.read(value)
134-
case _ => throw DocumentUnreadable(Messages.corruptedEntity)
128+
case x if x == classOf[WhiskAction] => WhiskAction.serdes.read(value)
129+
case x if x == classOf[WhiskActionMetaData] => WhiskActionMetaData.serdes.read(value)
130+
case x if x == classOf[WhiskPackage] => WhiskPackage.serdes.read(value)
131+
case x if x == classOf[WhiskActivation] => WhiskActivation.serdes.read(value)
132+
case x if x == classOf[WhiskTrigger] => WhiskTrigger.serdes.read(value)
133+
case x if x == classOf[WhiskRule] => WhiskRule.serdes.read(value)
134+
case x if x == classOf[WhiskActionDefaultVersion] => WhiskActionDefaultVersion.serdes.read(value)
135+
case _ => throw DocumentUnreadable(Messages.corruptedEntity)
135136
}
136137
value.asJsObject.fields.get("entityType").foreach {
137138
case JsString(entityType) if (doc.entityType != entityType) =>

‎core/controller/src/main/scala/org/apache/openwhisk/core/controller/Actions.scala‎

Lines changed: 105 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -220,53 +220,80 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
220220
* - 500 Internal Server Error
221221
*/
222222
override def create(user: Identity, entityName: FullyQualifiedEntityName)(implicit transid: TransactionId) = {
223-
parameter('overwrite ? false, 'deleteOld ? false) { (overwrite, deleteOld) =>
224-
entity(as[WhiskActionPut]) { content =>
225-
val request = content.resolve(user.namespace)
226-
val checkAdditionalPrivileges = entitleReferencedEntities(user, Privilege.READ, request.exec).flatMap {
227-
case _ => entitlementProvider.check(user, content.exec)
228-
}
229-
230-
onComplete(checkAdditionalPrivileges) {
231-
case Success(_) =>
232-
onComplete(WhiskActionVersionList.get(entityName, entityStore, false)) {
233-
case Success(result) if (result.versions.size >= actionMaxVersionLimit && !deleteOld) =>
234-
terminate(
235-
Forbidden,
236-
s"[PUT] entity has ${result.versions.size} versions exist which exceed $actionMaxVersionLimit, delete one of them before create new one or pass deleteOld=true to delete oldest version automatically")
237-
case Success(result) =>
238-
val id = result.matchedDocId(None).getOrElse(entityName.toDocId)
239-
putEntity(
240-
WhiskAction,
241-
entityStore,
242-
id,
243-
true,
244-
update(user, request) _,
245-
() => {
246-
make(user, entityName, request)
247-
},
248-
postProcess = Some { action: WhiskAction =>
249-
// delete oldest version when created successfully
250-
if (result.versions.size >= actionMaxVersionLimit) {
251-
val id = result.versions.minBy(_._1)._2
252-
WhiskAction.get(entityStore, DocId(id)) flatMap { entity =>
253-
WhiskAction.del(entityStore, DocInfo ! (id, entity.rev.rev)).map(_ => entity)
254-
} andThen {
255-
case _ =>
256-
WhiskActionVersionList.deleteCache(entityName)
257-
}
258-
} else {
223+
parameter('overwrite ? false, 'deleteOld ? false, 'defaultVersion.as[String] ? "") {
224+
(overwrite, deleteOld, defaultVersion) =>
225+
entity(as[WhiskActionPut]) { content =>
226+
Try {
227+
SemVer(defaultVersion)
228+
} match {
229+
case Success(version) =>
230+
onComplete(WhiskActionVersionList.get(entityName, entityStore, false)) {
231+
case Success(result) if (result.versions.keys.toVector.contains(version)) =>
232+
val dv = WhiskActionDefaultVersion(entityName.path, entityName.name, Some(version))
233+
putEntity(
234+
WhiskActionDefaultVersion,
235+
entityStore,
236+
dv.docid,
237+
true,
238+
(old: WhiskActionDefaultVersion) =>
239+
Future.successful(dv.revision[WhiskActionDefaultVersion](old.rev)),
240+
() => Future.successful(dv),
241+
postProcess = Some { version: WhiskActionDefaultVersion =>
259242
WhiskActionVersionList.deleteCache(entityName)
260-
}
261-
complete(OK, action)
262-
})
263-
case Failure(f) =>
264-
terminate(InternalServerError)
265-
}
266-
case Failure(f) =>
267-
super.handleEntitlementFailure(f)
243+
complete(OK, version)
244+
})
245+
case Success(_) =>
246+
terminate(Forbidden, s"[PUT] entity doesn't has version $version")
247+
case Failure(_) =>
248+
terminate(InternalServerError)
249+
}
250+
case Failure(_) =>
251+
val request = content.resolve(user.namespace)
252+
val checkAdditionalPrivileges = entitleReferencedEntities(user, Privilege.READ, request.exec).flatMap {
253+
case _ => entitlementProvider.check(user, content.exec)
254+
}
255+
256+
onComplete(checkAdditionalPrivileges) {
257+
case Success(_) =>
258+
onComplete(WhiskActionVersionList.get(entityName, entityStore, false)) {
259+
case Success(result) if (result.versions.size >= actionMaxVersionLimit && !deleteOld) =>
260+
terminate(
261+
Forbidden,
262+
s"[PUT] entity has ${result.versions.size} versions exist which exceed $actionMaxVersionLimit, delete one of them before create new one or pass deleteOld=true to delete oldest version automatically")
263+
case Success(result) =>
264+
val id = result.matchedDocId(None).getOrElse(entityName.toDocId)
265+
putEntity(
266+
WhiskAction,
267+
entityStore,
268+
id,
269+
true,
270+
update(user, request) _,
271+
() => {
272+
make(user, entityName, request)
273+
},
274+
postProcess = Some { action: WhiskAction =>
275+
// delete oldest version when created successfully
276+
if (result.versions.size >= actionMaxVersionLimit) {
277+
val id = result.versions.minBy(_._1)._2
278+
WhiskAction.get(entityStore, DocId(id)) flatMap { entity =>
279+
WhiskAction.del(entityStore, DocInfo ! (id, entity.rev.rev)).map(_ => entity)
280+
} andThen {
281+
case _ =>
282+
WhiskActionVersionList.deleteCache(entityName)
283+
}
284+
} else {
285+
WhiskActionVersionList.deleteCache(entityName)
286+
}
287+
complete(OK, action)
288+
})
289+
case Failure(f) =>
290+
terminate(InternalServerError)
291+
}
292+
case Failure(f) =>
293+
super.handleEntitlementFailure(f)
294+
}
295+
}
268296
}
269-
}
270297
}
271298
}
272299

@@ -388,10 +415,14 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
388415
docId,
389416
(a: WhiskAction) => Future.successful({}),
390417
postProcess = Some { action: WhiskAction =>
418+
// when default version is deleted or all versions are deleted, delete the default version entity
419+
if (version == results.defaultVersion || results.versions.size == 1)
420+
deleteDefaultVersion(
421+
WhiskActionDefaultVersion(entityName.path, entityName.name, results.defaultVersion))
391422
WhiskActionVersionList.deleteCache(entityName)
392423
complete(OK, action)
393424
})
394-
case None if (!deleteAll && results.versions.size > 1) =>
425+
case None if !deleteAll && results.versions.size > 1 =>
395426
terminate(
396427
Forbidden,
397428
s"[DEL] entity version not provided, you need to specify deleteAll=true to delete all versions for action $entityName")
@@ -412,6 +443,10 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
412443
case _ =>
413444
WhiskActionVersionList
414445
.deleteCache(entityName) // invalidate version list cache after all deletion completed
446+
deleteDefaultVersion(WhiskActionDefaultVersion(
447+
entityName.path,
448+
entityName.name,
449+
results.defaultVersion)) // delete default version entity since all versions are deleted
415450
}
416451

417452
onComplete(deleteFuture) {
@@ -443,6 +478,19 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
443478
}
444479
}
445480

481+
private def deleteDefaultVersion(defaultVersion: WhiskActionDefaultVersion)(implicit transid: TransactionId): Unit = {
482+
WhiskActionDefaultVersion.get(entityStore, defaultVersion.docid) map { entity =>
483+
WhiskActionDefaultVersion.del(entityStore, defaultVersion.docid.asDocInfo(entity.rev)) andThen {
484+
case Success(_) =>
485+
logging.info(this, s"[DEL] default version for ${defaultVersion.fullyQualifiedName(false)} is deleted")
486+
case Failure(t) =>
487+
logging.error(
488+
this,
489+
s"[DEL] failed to delete default version for ${defaultVersion.fullyQualifiedName(false)}, error: $t")
490+
}
491+
}
492+
}
493+
446494
/** Checks for package binding case. we don't want to allow get for a package binding in shared package */
447495
private def fetchEntity(entityName: FullyQualifiedEntityName,
448496
env: Option[Parameters],
@@ -499,10 +547,17 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
499547
*/
500548
override def fetch(user: Identity, entityName: FullyQualifiedEntityName, env: Option[Parameters])(
501549
implicit transid: TransactionId) = {
502-
parameter('code ? true, 'version.as[SemVer] ?) { (code, version) =>
503-
//check if execute only is enabled, and if there is a discrepancy between the current user's namespace
504-
//and that of the entity we are trying to fetch
505-
if (executeOnly && user.namespace.name != entityName.namespace) {
550+
parameter('code ? true, 'version.as[SemVer] ?, 'showVersions ? false) { (code, version, showVersions) =>
551+
if (showVersions) {
552+
onComplete(WhiskActionVersionList.get(entityName, entityStore)) {
553+
case Success(res) =>
554+
complete(OK, res)
555+
case Failure(t) =>
556+
terminate(Forbidden, forbiddenGetAction(entityName.path.asString))
557+
}
558+
//check if execute only is enabled, and if there is a discrepancy between the current user's namespace
559+
//and that of the entity we are trying to fetch
560+
} else if (executeOnly && user.namespace.name != entityName.namespace) {
506561
terminate(Forbidden, forbiddenGetAction(entityName.path.asString))
507562
} else {
508563
fetchEntity(entityName, env, code, version)

‎core/controller/src/main/scala/org/apache/openwhisk/core/controller/Packages.scala‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,10 @@ trait WhiskPackagesApi extends WhiskCollectionAPI with ReferencedEntities {
162162
case _ =>
163163
list.foreach { action =>
164164
WhiskActionVersionList.deleteCache(action.fullyQualifiedName(false))
165+
val version = WhiskActionDefaultVersion(action.namespace, action.name, None)
166+
WhiskActionDefaultVersion.get(entityStore, version.docid) foreach { versionWithRevision =>
167+
WhiskActionDefaultVersion.del(entityStore, versionWithRevision.docinfo)
168+
}
165169
}
166170
} flatMap { _ =>
167171
Future.successful({})

‎tests/src/test/scala/common/WskCliOperations.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -206,6 +206,8 @@ class CliActionOperations(override val wsk: RunCliCmd)
206206
update: Boolean = false,
207207
web: Option[String] = None,
208208
websecure: Option[String] = None,
209+
deleteOld: Boolean = true,
210+
defaultVersion: Option[String] = None,
209211
expectedExitCode: Int = SUCCESS_EXIT)(implicit wp: WskProps): RunResult = {
210212
val params = Seq(noun, if (!update) "create" else "update", "--auth", wp.authKey, fqn(name)) ++ {
211213
artifact map { Seq(_) } getOrElse Seq.empty

‎tests/src/test/scala/common/WskOperations.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -245,6 +245,8 @@ trait ActionOperations extends DeleteFromCollectionOperations with ListOrGetFrom
245245
update: Boolean = false,
246246
web: Option[String] = None,
247247
websecure: Option[String] = None,
248+
deleteOld: Boolean = true,
249+
defaultVersion: Option[String] = None,
248250
expectedExitCode: Int = SUCCESS_EXIT)(implicit wp: WskProps): RunResult
249251

250252
def invoke(name: String,

0 commit comments

Comments
 (0)