You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openwhisk.apache.org by ra...@apache.org on 2018/02/15 23:31:43 UTC
[incubator-openwhisk] branch master updated: Support action
continuations in the controller (#3202)
This is an automated email from the ASF dual-hosted git repository.
rabbah 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 78105f9 Support action continuations in the controller (#3202)
78105f9 is described below
commit 78105f92dbb0ac61cda15cb046779cfbc39d8791
Author: Olivier Tardieu <ta...@users.noreply.github.com>
AuthorDate: Thu Feb 15 18:31:40 2018 -0500
Support action continuations in the controller (#3202)
This commit introduces a new type of actions called conductors. Conductors are a kind of sequence with dynamically computed steps. While the components of a sequence must be specified before invocation, components of a conductor can be decided as the composition is running.
Like the sequence implementation, the bulk of the implementation for supporting conductors lies in the controller. It chains the invocations of the component actions. It keeps track of the execution state. It produces the final activation record for the top level conductor invocation.
In contrast with sequences and in an attempt to keep the changes contained in the controller code, a conductor is not a new kind of action but rather a action with a marker annotation. This means no CLI changes for instance.
A conductor action is an action with a defined "conductor" annotation. The value of the annotation must be a truthy to have effect. The annotation is ignored on sequences.
Conductor actions drive the execution of compositions (functions invoked one after the other). A conductor action may either return a final result or a triplet { action, params, state }. In the latter case, the specified component action is invoked on the specified params object. Upon completion of this action the conductor action is reinvoked with a payload that combines the output of the action with the state returned by the previous conductor invocation. The composition result is t [...]
A composition is therefore a sequence of invocations alternating invocations of the (fixed) conductor action with invocations of the dynamically specified components of the composition. In a sense, the conductor action acts as a scheduler that decodes and executes a sequential program. Conductor actions make it possible to support dynamic composition in OpenWhisk without committing to any particular execution model (e.g., finite state machines of one kind or another), or invocation co [...]
Compositions may be nested, i.e, a step in a composition may itself be a composition. The controller recognizes this pattern and handles it efficiently.
Like a sequence, one composition is materialized by one action. Creating a composition requires the definition of one conductor action. Deleting the composition consists in deleting the corresponding action. Like a sequence, a composition refers to the composed actions by name (i.e., by reference).
There is no distinction between invoking a conductor and an action. Both invocations returns an activation id that can be used to fetch the result of the execution of the action or composition. Both invocations can be blocking. Both can be web actions.
As for sequences, there is a configurable upper bound on the length of a composition (number of steps). In contrast with sequences, the limit is assessed at run time, and the composition invocation is aborted if the limit is exceeded. An entitlement check verifies that each composed action is accessible prior to its execution. A composition invocation like a sequence invocation counts as one invocation viz. throttling. In particular, a composition will not be aborted during execution [...]
This commit comes with unit tests for the controller as well as integration tests.
Example:
$ cat increment.js
function main({ value }) { return { value: value + 1 } }
$ cat conductor.js
function main(params) {
switch (params.$step || 0) {
case 0: delete params.$step; return { params, action: 'increment', state: { $step: 1 } }
case 1: delete params.$step; return { params, action: 'increment', state: { $step: 2 } }
case 2: delete params.$step; return params
}
}
$ wsk action update increment increment.js
$ wsk action update conductor conductor.js -a conductor true
$ wsk action invoke conductor -r -p value 3
{
"value": 5
}
This example demonstrates a simple conductor with two increment steps. The progress of the execution is tracked via the $step field of the parameter object that is propagated from a conductor invocation to the next but hidden from the composed actions.
---
.../main/scala/whisk/core/entity/Parameter.scala | 11 +
.../scala/whisk/core/entity/WhiskActivation.scala | 7 +
.../src/main/scala/whisk/http/ErrorResponse.scala | 12 +
.../core/controller/actions/PrimitiveActions.scala | 411 ++++++++++++++++++++-
docs/conductors.md | 253 +++++++++++++
tests/dat/actions/conductor.js | 15 +
tests/dat/actions/step.js | 7 +
.../scala/system/basic/WskConductorTests.scala | 330 +++++++++++++++++
.../scala/system/basic/WskRestConductorTests.scala | 28 ++
.../core/controller/test/ConductorsApiTests.scala | 374 +++++++++++++++++++
10 files changed, 1446 insertions(+), 2 deletions(-)
diff --git a/common/scala/src/main/scala/whisk/core/entity/Parameter.scala b/common/scala/src/main/scala/whisk/core/entity/Parameter.scala
index 937913d..0dd6518 100644
--- a/common/scala/src/main/scala/whisk/core/entity/Parameter.scala
+++ b/common/scala/src/main/scala/whisk/core/entity/Parameter.scala
@@ -101,6 +101,17 @@ protected[core] class Parameters protected[entity] (private val params: Map[Para
/** Retrieves parameter by name if it exists. Returns that parameter if it is deserializable to {@code T} */
protected[core] def getAs[T: JsonReader](p: String): Option[T] = get(p).flatMap(js => Try(js.convertTo[T]).toOption)
+
+ /** Retrieves parameter by name if it exist. Returns true if parameter exists and has truthy value. */
+ protected[core] def isTruthy(p: String): Boolean = {
+ get(p) map {
+ case JsBoolean(b) => b
+ case JsNumber(n) => n != 0
+ case JsString(s) => s.nonEmpty
+ case JsNull => false
+ case _ => true
+ } getOrElse false
+ }
}
/**
diff --git a/common/scala/src/main/scala/whisk/core/entity/WhiskActivation.scala b/common/scala/src/main/scala/whisk/core/entity/WhiskActivation.scala
index f6026ef..8c18a98 100644
--- a/common/scala/src/main/scala/whisk/core/entity/WhiskActivation.scala
+++ b/common/scala/src/main/scala/whisk/core/entity/WhiskActivation.scala
@@ -135,6 +135,13 @@ object WhiskActivation
val causedByAnnotation = "causedBy"
val initTimeAnnotation = "initTime"
val waitTimeAnnotation = "waitTime"
+ val conductorAnnotation = "conductor"
+
+ /** Some field names for compositions */
+ val actionField = "action"
+ val paramsField = "params"
+ val stateField = "state"
+ val valueField = "value"
protected[entity] implicit val instantSerdes = new RootJsonFormat[Instant] {
def write(t: Instant) = t.toEpochMilli.toJson
diff --git a/common/scala/src/main/scala/whisk/http/ErrorResponse.scala b/common/scala/src/main/scala/whisk/http/ErrorResponse.scala
index bd0bfb3..f161618 100644
--- a/common/scala/src/main/scala/whisk/http/ErrorResponse.scala
+++ b/common/scala/src/main/scala/whisk/http/ErrorResponse.scala
@@ -116,6 +116,18 @@ object Messages {
s"Timeout reached when retrieving activation $id for sequence component."
val sequenceActivationFailure = "Sequence failed."
+ /** Error messages for compositions. */
+ val compositionIsTooLong = "Too many actions in the composition."
+ val compositionActivationFailure = "Activation failure during composition."
+ def compositionActivationTimeout(id: ActivationId) =
+ s"Timeout reached when retrieving activation $id during composition."
+ def compositionComponentInvalid(value: JsValue) =
+ s"Failed to parse action name from json value $value during composition."
+ def compositionComponentNotFound(name: String) =
+ s"""Failed to resolve action with name "$name" during composition."""
+ def compositionComponentNotAccessible(name: String) =
+ s"""Failed entitlement check for action with name "$name" during composition."""
+
/** Error messages for bad requests where parameters do not conform. */
val parametersNotAllowed = "Request defines parameters that are not allowed (e.g., reserved properties)."
def invalidTimeout(max: FiniteDuration) = s"Timeout must be number of milliseconds up to ${max.toMillis}."
diff --git a/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala b/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
index 2d23b86..2006a6b 100644
--- a/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
+++ b/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
@@ -17,11 +17,18 @@
package whisk.core.controller.actions
+import java.time.Clock
+import java.time.Instant
+
+import scala.collection.mutable.Buffer
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.Promise
import scala.concurrent.duration._
+import scala.language.postfixOps
import scala.util.Failure
+import scala.util.Success
+import scala.util.Try
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
@@ -35,9 +42,13 @@ import whisk.common.TransactionId
import whisk.core.connector.ActivationMessage
import whisk.core.controller.WhiskServices
import whisk.core.database.NoDocumentException
+import whisk.core.entitlement._
+import whisk.core.entitlement.Resource
import whisk.core.entity._
+import whisk.core.entity.size.SizeInt
import whisk.core.entity.types.ActivationStore
import whisk.core.entity.types.EntityStore
+import whisk.http.Messages._
import whisk.utils.ExecutionContextFactory.FutureExtensions
import akka.event.Logging.InfoLevel
@@ -65,18 +76,67 @@ protected[actions] trait PrimitiveActions {
/** Database service to get activations. */
protected val activationStore: ActivationStore
+ /** A method that knows how to invoke a sequence of actions. */
+ protected[actions] def invokeSequence(
+ user: Identity,
+ action: WhiskActionMetaData,
+ components: Vector[FullyQualifiedEntityName],
+ payload: Option[JsObject],
+ waitForOutermostResponse: Option[FiniteDuration],
+ cause: Option[ActivationId],
+ topmost: Boolean,
+ atomicActionsCount: Int)(implicit transid: TransactionId): Future[(Either[ActivationId, WhiskActivation], Int)]
+
/**
+ * A method that knows how to invoke a single primitive action or a composition.
+ *
+ * A composition is a kind of sequence of actions that is dynamically computed.
+ * The execution of a composition is triggered by the invocation of a conductor action.
+ * A conductor action is an executable action with a truthy "conductor" annotation.
+ * Sequences cannot be compositions: the "conductor" annotation on a sequence has no effect.
+ *
+ * A conductor action may either return a final result or a triplet { action, params, state }.
+ * In the latter case, the specified component action is invoked on the specified params object.
+ * Upon completion of this action the conductor action is reinvoked with a payload that combines
+ * the output of the action with the state returned by the previous conductor invocation.
+ * The composition result is the result of the final conductor invocation in the chain of invocations.
+ *
+ * The trace of a composition obeys the grammar: conductorInvocation(componentInvocation conductorInvocation)*
+ *
+ * The activation records for a composition and its components mimic the activation records of sequences.
+ * They include the same "topmost", "kind", and "causedBy" annotations with the same semantics.
+ * The activation record for a composition also includes a specific annotation "conductor" with value true.
+ */
+ protected[actions] def invokeSingleAction(
+ user: Identity,
+ action: ExecutableWhiskActionMetaData,
+ payload: Option[JsObject],
+ waitForResponse: Option[FiniteDuration],
+ cause: Option[ActivationId])(implicit transid: TransactionId): Future[Either[ActivationId, WhiskActivation]] = {
+
+ if (action.annotations.isTruthy(WhiskActivation.conductorAnnotation)) {
+ invokeComposition(user, action, payload, waitForResponse, cause)
+ } else {
+ invokeSimpleAction(user, action, payload, waitForResponse, cause)
+ }
+ }
+
+ /**
+ * A method that knows how to invoke a single primitive action.
+ *
* Posts request to the loadbalancer. If the loadbalancer accepts the requests with an activation id,
* then wait for the result of the activation if necessary.
*
* NOTE:
- * For activations of actions, cause is populated only for actions that were invoked as a result of a sequence activation.
+ * Cause is populated only for actions that were invoked as a result of a sequence activation or a composition.
* For actions that are enclosed in a sequence and are activated as a result of the sequence activation, the cause
* contains the activation id of the immediately enclosing sequence.
* e.g.,: s -> a, x, c and x -> c (x and s are sequences, a, b, c atomic actions)
* cause for a, x, c is the activation id of s
* cause for c is the activation id of x
* cause for s is not defined
+ * For actions that are enclosed in a composition and are activated as a result of the composition activation,
+ * the cause contains the activation id of the immediately enclosing composition.
*
* @param user the identity invoking the action
* @param action the action to invoke
@@ -90,7 +150,7 @@ protected[actions] trait PrimitiveActions {
* or these custom failures:
* RequestEntityTooLarge if the message is too large to to post to the message bus
*/
- protected[actions] def invokeSingleAction(
+ private def invokeSimpleAction(
user: Identity,
action: ExecutableWhiskActionMetaData,
payload: Option[JsObject],
@@ -141,6 +201,350 @@ protected[actions] trait PrimitiveActions {
}
/**
+ * Mutable cumulative accounting of what happened during the execution of a composition.
+ *
+ * Compositions are aborted if the number of action invocations exceeds a limit.
+ * The permitted max is n component invocations plus 2n+1 conductor invocations (where n is the actionSequenceLimit).
+ * The max is chosen to permit a sequence with up to n primitive actions.
+ *
+ * NOTE:
+ * A sequence invocation counts as one invocation irrespective of the number of action invocations in the sequence.
+ * If one component of a composition is also a composition, the caller and callee share the same accounting object.
+ * The counts are shared between callers and callees so the limit applies globally.
+ *
+ * @param components the current count of component actions already invoked
+ * @param conductors the current count of conductor actions already invoked
+ */
+ private case class CompositionAccounting(var components: Int = 0, var conductors: Int = 0)
+
+ /**
+ * A mutable session object to keep track of the execution of one composition.
+ *
+ * NOTE:
+ * The session object is not shared between callers and callees.
+ *
+ * @param activationId the activationId for the composition (ie the activation record for the composition)
+ * @param start the start time for the composition
+ * @param action the conductor action responsible for the execution of the composition
+ * @param cause the cause of the composition (activationId of the enclosing sequence or composition if any)
+ * @param duration the "user" time so far executing the composition (sum of durations for
+ * all actions invoked so far which is different from the total time spent executing the composition)
+ * @param maxMemory the maximum memory annotation observed so far for the conductor action and components
+ * @param state the json state object to inject in the parameter object of the next conductor invocation
+ * @param accounting the global accounting object used to abort compositions requiring too many action invocations
+ * @param logs a mutable buffer that is appended with new activation ids as the composition unfolds
+ * (in contrast with sequences, the logs of a hierarchy of compositions is not flattened)
+ */
+ private case class Session(activationId: ActivationId,
+ start: Instant,
+ action: ExecutableWhiskActionMetaData,
+ cause: Option[ActivationId],
+ var duration: Long,
+ var maxMemory: ByteSize,
+ var state: Option[JsObject],
+ accounting: CompositionAccounting,
+ logs: Buffer[ActivationId])
+
+ /**
+ * A method that knows how to invoke a composition.
+ *
+ * The method instantiates the session object for the composition and invokes the conductor action.
+ * It waits for the activation response, synthesizes the activation record and writes it to the datastore.
+ * It distinguishes nested, blocking and non-blocking invokes, returning either the activation or the activation id.
+ *
+ * @param user the identity invoking the action
+ * @param action the conductor action to invoke for the composition
+ * @param payload the dynamic arguments for the activation
+ * @param waitForResponse if not empty, wait upto specified duration for a response (this is used for blocking activations)
+ * @param cause the activation id that is responsible for this invoke/activation
+ * @param accounting the accounting object for the caller if any
+ * @param transid a transaction id for logging
+ * @return a promise that completes with one of the following successful cases:
+ * Right(WhiskActivation) if waiting for a response and response is ready within allowed duration,
+ * Left(ActivationId) if not waiting for a response, or allowed duration has elapsed without a result ready
+ */
+ private def invokeComposition(user: Identity,
+ action: ExecutableWhiskActionMetaData,
+ payload: Option[JsObject],
+ waitForResponse: Option[FiniteDuration],
+ cause: Option[ActivationId],
+ accounting: Option[CompositionAccounting] = None)(
+ implicit transid: TransactionId): Future[Either[ActivationId, WhiskActivation]] = {
+
+ val session = Session(
+ activationId = activationIdFactory.make(),
+ start = Instant.now(Clock.systemUTC()),
+ action,
+ cause,
+ duration = 0,
+ maxMemory = action.limits.memory.megabytes MB,
+ state = None,
+ accounting = accounting.getOrElse(CompositionAccounting()), // share accounting with caller
+ logs = Buffer.empty)
+
+ val response: Future[Either[ActivationId, WhiskActivation]] =
+ invokeConductor(user, payload, session).map(response => Right(completeActivation(user, session, response)))
+
+ // is caller waiting for the result of the activation?
+ cause
+ .map(_ => response) // ignore waitForResponse when not topmost
+ .orElse(
+ // blocking invoke, wait until timeout
+ waitForResponse.map(response.withAlternativeAfterTimeout(_, Future.successful(Left(session.activationId)))))
+ .getOrElse(
+ // no, return the session id
+ Future.successful(Left(session.activationId)))
+ }
+
+ /**
+ * A method that knows how to handle a conductor invocation.
+ *
+ * This method prepares the payload and invokes the conductor action.
+ * It parses the result and extracts the name of the next component action if any.
+ * It either invokes the desired component action or completes the composition invocation.
+ * It also checks the invocation counts against the limits.
+ *
+ * @param user the identity invoking the action
+ * @param payload the dynamic arguments for the activation
+ * @param session the session object for this composition
+ * @param transid a transaction id for logging
+ */
+ private def invokeConductor(user: Identity, payload: Option[JsObject], session: Session)(
+ implicit transid: TransactionId): Future[ActivationResponse] = {
+
+ if (session.accounting.conductors > 2 * actionSequenceLimit) {
+ // composition is too long
+ Future.successful(ActivationResponse.applicationError(compositionIsTooLong))
+ } else {
+ // inject state into payload if any
+ val params = session.state
+ .map(state => Some(JsObject(payload.getOrElse(JsObject()).fields ++ state.fields)))
+ .getOrElse(payload)
+
+ // invoke conductor action
+ session.accounting.conductors += 1
+ val activationResponse =
+ invokeSimpleAction(
+ user,
+ action = session.action,
+ payload = params,
+ waitForResponse = Some(session.action.limits.timeout.duration + 1.minute), // wait for result
+ cause = Some(session.activationId)) // cause is session id
+
+ waitForActivation(user, session, activationResponse).flatMap {
+ case Left(response) => // unsuccessful invocation, return error response
+ Future.successful(response)
+ case Right(activation) => // successful invocation
+ val result = activation.resultAsJson
+
+ // extract params from result, auto boxing result if not a dictionary
+ val params = result.fields.get(WhiskActivation.paramsField).map {
+ case obj: JsObject => obj
+ case value => JsObject(WhiskActivation.valueField -> value)
+ }
+
+ // update session state, auto boxing state if not a dictionary
+ session.state = result.fields.get(WhiskActivation.stateField).map {
+ case obj: JsObject => obj
+ case value => JsObject(WhiskActivation.stateField -> value)
+ }
+
+ // extract next action from result and invoke
+ result.fields.get(WhiskActivation.actionField) match {
+ case None =>
+ // no next action, end composition execution, return to caller
+ Future.successful(ActivationResponse(activation.response.statusCode, Some(params.getOrElse(result))))
+ case Some(next) =>
+ Try(next.convertTo[EntityPath]) match {
+ case Failure(t) =>
+ // parsing failure
+ Future.successful(ActivationResponse.applicationError(compositionComponentInvalid(next)))
+ case Success(_) if session.accounting.components >= actionSequenceLimit =>
+ // composition is too long
+ Future.successful(ActivationResponse.applicationError(compositionIsTooLong))
+ case Success(next) =>
+ // resolve and invoke next action
+ val fqn = (if (next.defaultPackage) EntityPath.DEFAULT.addPath(next) else next)
+ .resolveNamespace(user.namespace)
+ .toFullyQualifiedEntityName
+ val resource = Resource(fqn.path, Collection(Collection.ACTIONS), Some(fqn.name.asString))
+ entitlementProvider
+ .check(user, Privilege.ACTIVATE, Set(resource), noThrottle = true)
+ .flatMap { _ =>
+ // successful entitlement check
+ WhiskActionMetaData
+ .resolveActionAndMergeParameters(entityStore, fqn)
+ .flatMap {
+ case next =>
+ // successful resolution
+ invokeComponent(user, action = next, payload = params, session)
+ }
+ .recover {
+ case _ =>
+ // resolution failure
+ ActivationResponse.applicationError(compositionComponentNotFound(next.asString))
+ }
+ }
+ .recover {
+ case _ =>
+ // failed entitlement check
+ ActivationResponse.applicationError(compositionComponentNotAccessible(next.asString))
+ }
+ }
+ }
+ }
+ }
+
+ }
+
+ /**
+ * A method that knows how to handle a component invocation.
+ *
+ * This method distinguishes primitive actions, sequences, and compositions.
+ * The conductor action is reinvoked after the successful invocation of the component.
+ *
+ * @param user the identity invoking the action
+ * @param action the component action to invoke
+ * @param payload the dynamic arguments for the activation
+ * @param session the session object for this composition
+ * @param transid a transaction id for logging
+ */
+ private def invokeComponent(user: Identity, action: WhiskActionMetaData, payload: Option[JsObject], session: Session)(
+ implicit transid: TransactionId): Future[ActivationResponse] = {
+
+ val exec = action.toExecutableWhiskAction
+ val activationResponse: Future[Either[ActivationId, WhiskActivation]] = exec match {
+ case Some(action) if action.annotations.isTruthy(WhiskActivation.conductorAnnotation) => // composition
+ // invokeComposition will increase the invocation counts
+ invokeComposition(
+ user,
+ action,
+ payload,
+ waitForResponse = None, // not topmost, hence blocking, no need for timeout
+ cause = Some(session.activationId),
+ accounting = Some(session.accounting))
+ case Some(action) => // primitive action
+ session.accounting.components += 1
+ invokeSimpleAction(
+ user,
+ action,
+ payload,
+ waitForResponse = Some(action.limits.timeout.duration + 1.minute),
+ cause = Some(session.activationId))
+ case None => // sequence
+ session.accounting.components += 1
+ val SequenceExecMetaData(components) = action.exec
+ invokeSequence(
+ user,
+ action,
+ components,
+ payload,
+ waitForOutermostResponse = None,
+ cause = Some(session.activationId),
+ topmost = false,
+ atomicActionsCount = 0).map(r => r._1)
+ }
+
+ waitForActivation(user, session, activationResponse).flatMap {
+ case Left(response) => // unsuccessful invocation, return error response
+ Future.successful(response)
+ case Right(activation) => // reinvoke conductor on component result
+ invokeConductor(user, payload = Some(activation.resultAsJson), session = session)
+ }
+ }
+
+ /**
+ * Waits for a response from a conductor of component action invocation.
+ * Handles internal errors (activation failure or timeout).
+ * Logs the activation id and updates the duration and max memory for the session.
+ * Returns the activation record if successful, the error response if not.
+ *
+ * @param user the identity invoking the action
+ * @param session the session object for this composition
+ * @param activationResponse the future activation to wait on
+ * @param transid a transaction id for logging
+ */
+ private def waitForActivation(user: Identity,
+ session: Session,
+ activationResponse: Future[Either[ActivationId, WhiskActivation]])(
+ implicit transid: TransactionId): Future[Either[ActivationResponse, WhiskActivation]] = {
+
+ activationResponse
+ .map {
+ case Left(activationId) => // invocation timeout
+ session.logs += activationId
+ Left(ActivationResponse.whiskError(compositionActivationTimeout(activationId)))
+ case Right(activation) => // successful invocation
+ session.logs += activation.activationId
+ // activation.duration should be defined but this is not reflected by the type so be defensive
+ // end - start is a sensible default but not the correct value for sequences and compositions
+ session.duration += activation.duration.getOrElse(activation.end.toEpochMilli - activation.start.toEpochMilli)
+ activation.annotations.get("limits").foreach { limitsAnnotation =>
+ limitsAnnotation.asJsObject.getFields("memory") match {
+ case Seq(JsNumber(memory)) =>
+ session.maxMemory = Math.max(session.maxMemory.toMB.toInt, memory.toInt) MB
+ }
+ }
+ Right(activation)
+ }
+ .recover { // invocation failure
+ case _ => Left(ActivationResponse.whiskError(compositionActivationFailure))
+ }
+ }
+
+ /**
+ * Creates an activation for a composition and writes it back to the datastore.
+ * Returns the activation.
+ */
+ private def completeActivation(user: Identity, session: Session, response: ActivationResponse)(
+ implicit transid: TransactionId): WhiskActivation = {
+
+ // compute max memory
+ val sequenceLimits = Parameters(
+ WhiskActivation.limitsAnnotation,
+ ActionLimits(session.action.limits.timeout, MemoryLimit(session.maxMemory), session.action.limits.logs).toJson)
+
+ // set causedBy if not topmost
+ val causedBy = session.cause.map { _ =>
+ Parameters(WhiskActivation.causedByAnnotation, JsString(Exec.SEQUENCE))
+ }
+
+ val end = Instant.now(Clock.systemUTC())
+
+ // create the whisk activation
+ val activation = WhiskActivation(
+ namespace = user.namespace.toPath,
+ name = session.action.name,
+ user.subject,
+ activationId = session.activationId,
+ start = session.start,
+ end = end,
+ cause = session.cause,
+ response = response,
+ logs = ActivationLogs(session.logs.map(_.asString).toVector),
+ version = session.action.version,
+ publish = false,
+ annotations = Parameters(WhiskActivation.topmostAnnotation, JsBoolean(session.cause.isEmpty)) ++
+ Parameters(WhiskActivation.pathAnnotation, JsString(session.action.fullyQualifiedName(false).asString)) ++
+ Parameters(WhiskActivation.kindAnnotation, JsString(Exec.SEQUENCE)) ++
+ Parameters(WhiskActivation.conductorAnnotation, JsBoolean(true)) ++
+ causedBy ++
+ sequenceLimits,
+ duration = Some(session.duration))
+
+ logging.debug(this, s"recording activation '${activation.activationId}'")
+ WhiskActivation.put(activationStore, activation)(transid, notifier = None) onComplete {
+ case Success(id) => logging.debug(this, s"recorded activation")
+ case Failure(t) =>
+ logging.error(
+ this,
+ s"failed to record activation ${activation.activationId} with error ${t.getLocalizedMessage}")
+ }
+
+ activation
+ }
+
+ /**
* Waits for a response from the message bus (e.g., Kafka) containing the result of the activation. This is the fast path
* used for blocking calls where only the result of the activation is needed. This path is called active acknowledgement
* or active ack.
@@ -189,6 +593,9 @@ protected[actions] trait PrimitiveActions {
}
})
}
+
+ /** Max atomic action count allowed for sequences */
+ private lazy val actionSequenceLimit = whiskConfig.actionSequenceLimit.toInt
}
/** Companion to the ActivationFinisher. */
diff --git a/docs/conductors.md b/docs/conductors.md
new file mode 100644
index 0000000..8ab888f
--- /dev/null
+++ b/docs/conductors.md
@@ -0,0 +1,253 @@
+# Conductor Actions
+
+Conductor actions make it possible to build and invoke a series of actions, similar to sequences. However, whereas the components of a sequence action must be specified before invoking the sequence, conductor actions can decide the series of actions to invoke at run time.
+
+In this document, we specify conductor actions and illustrate them with a simple example: a _tripleAndIncrement_ action.
+
+Suppose we define a _triple_ action in a source file `triple.js`:
+
+```javascript
+function main({ value }) { return { value: value * 3 } }
+```
+
+We create the action _triple_:
+
+```
+wsk action create triple triple.js
+```
+
+We define an _increment_ action in a source file `increment.js`:
+
+```javascript
+function main({ value }) { return { value: value + 1 } }
+```
+
+We create the action _increment_:
+
+```
+wsk action create increment increment.js
+```
+
+## Conductor annotation
+
+We define the _tripleAndIncrement_ action in a source file `tripleAndIncrement.js`:
+
+```javascript
+function main(params) {
+ let step = params.$step || 0
+ delete params.$step
+ switch (step) {
+ case 0: return { action: 'triple', params, state: { $step: 1 } }
+ case 1: return { action: 'increment', params, state: { $step: 2 } }
+ case 2: return { params }
+ }
+}
+```
+
+We create a _conductor_ action by specifying the _conductor_ annotation:
+
+```
+wsk action create tripleAndIncrement tripleAndIncrement.js -a conductor true
+```
+
+A _conductor action_ is an action with a _conductor_ annotation with a value that is not _falsy_, i.e., a value that is different from zero, null, false, and the empty string.
+
+At this time, the conductor annotation is ignored on sequence actions.
+
+Because a conductor action is an action, it has all the attributes of an action (name, namespace, default parameters, limits...) and it can be managed as such, for instance using the `wsk action` CLI commands. It can be part of a package or be a web action.
+
+In essence, the _tripleAndIncrement_ action builds a sequence of two actions by encoding a program with three steps:
+
+- step 0: invoke the _triple_ action on the input dictionary,
+- step 1: invoke the _increment_ action on the output dictionary from step 1,
+- step 2: return the output dictionary from step 2.
+
+At each step, the conductor action specifies how to continue or terminate the execution by means of a _continuation_. We explain continuations after discussing invocation and activations.
+
+## Invocation
+
+A conductor action is invoked like a regular [action](actions.md), for instance:
+
+```
+wsk action invoke tripleAndIncrement -r -p value 3
+```
+```json
+{
+ "value": 10
+}
+```
+
+Blocking and non-blocking invocations are supported. As usual, a blocking invocation may timeout before the completion of the invocation.
+
+## Activations
+
+One invocation of the conductor action results in multiple activations, for instance:
+
+```
+wsk action invoke quadruple -p value 3
+```
+```
+ok: invoked /_/quadruple with id 4f91f9ed0d874aaa91f9ed0d87baaa07
+```
+```
+wsk activation list
+```
+```
+activations
+fd89b99a90a1462a89b99a90a1d62a8e tripleAndIncrement
+eaec119273d94087ac119273d90087d0 increment
+3624ad829d4044afa4ad829d40e4af60 tripleAndIncrement
+a1f58ade9b1e4c26b58ade9b1e4c2614 triple
+3624ad829d4044afa4ad829d40e4af60 tripleAndIncrement
+4f91f9ed0d874aaa91f9ed0d87baaa07 tripleAndIncrement
+```
+
+There are six activation records in this example, one matching the activation id returned on invocation (`4f91f9ed0d874aaa91f9ed0d87baaa07`) plus five additional records for activations _caused_ by this invocation. The _primary_ activation record is the last one in the list because it has the earliest start time.
+
+The five additional activations are:
+
+- one activation of the _triple_ action with input `{ value: 3 }` and output `{ value: 9 }`,
+- one activation of the _increment_ action with input `{ value: 9 }` and output `{ value: 10 }`,
+- three _secondary_ activations of the _tripleAndIncrement_ action.
+
+### Causality
+
+We say the invocation of the conductor action is the _cause_ of _component_ action invocations as well as _secondary_ activations of the conductor action. These activations are _derived_ activations.
+
+The cause field of the _derived_ activation records is set to the id for the _primary_ activation record.
+
+### Primary activations
+
+The primary activation record for the invocation of a conductor action is a synthetic record similar to the activation record of a sequence action. The primary activation record summarizes the series of derived activations:
+
+- its result is the result of the last action in the series (possibly unboxed, see below),
+- its logs are the ordered list of component and secondary activations,
+- its duration is the sum of the durations of these activations,
+- its start time is less or equal to the start time of the first derived activation in the series,
+- its end time is greater or equal to the end time of the last derived activation in the series.
+
+```
+wsk activation get 4f91f9ed0d874aaa91f9ed0d87baaa07
+```
+```
+ok: got activation 4f91f9ed0d874aaa91f9ed0d87baaa07
+{
+ "namespace": "guest",
+ "name": "composition",
+ "version": "0.0.1",
+ "subject": "guest",
+ "activationId": "4f91f9ed0d874aaa91f9ed0d87baaa07",
+ "start": 1516379705819,
+ "end": 1516379707803,
+ "duration": 457,
+ "response": {
+ "status": "success",
+ "statusCode": 0,
+ "success": true,
+ "result": {
+ "value": 12
+ }
+ },
+ "logs": [
+ "3624ad829d4044afa4ad829d40e4af60",
+ "a1f58ade9b1e4c26b58ade9b1e4c2614",
+ "47cb5aa5e4504f818b5aa5e450ef810f",
+ "eaec119273d94087ac119273d90087d0",
+ "fd89b99a90a1462a89b99a90a1d62a8e"
+ ],
+ "annotations": [
+ {
+ "key": "topmost",
+ "value": true
+ },
+ {
+ "key": "path",
+ "value": "guest/tripleAndIncrement"
+ },
+ {
+ "key": "conductor",
+ "value": true
+ },
+ {
+ "key": "kind",
+ "value": "sequence"
+ },
+ {
+ "key": "limits",
+ "value": {
+ "logs": 10,
+ "memory": 256,
+ "timeout": 60000
+ }
+ }
+ "publish": false
+}
+```
+
+If a component action itself is a sequence or conductor action, the logs contain only the id for the component activation. They do not contain the ids for the activations caused by this component. This is different from nested sequence actions.
+
+### Secondary activations
+
+The secondary activations of the conductor action are responsible for orchestrating the invocations of the component actions.
+
+An invocation of a conductor action starts with a secondary activation and alternates secondary activations of this conductor action with invocations of the component actions. It normally ends with a secondary activation of the conductor action. In our example, the five derived activations are interleaved as follows:
+
+ 1. secondary _tripleAndIncrement_ activation,
+ 2. _triple_ activation,
+ 3. secondary _tripleAndIncrement_ activation,
+ 4. _increment_ activation,
+ 5. secondary _tripleAndIncrement_ activation.
+
+Intuitively, secondary activations of the conductor action decide which component actions to invoke by running before, in-between, and after the component actions. In this example, the _tripleAndIncrement_ main function runs three times.
+
+Only an internal error (invocation failure or timeout) may result in an even number of derived activations.
+
+### Annotations
+
+Primary activation records include the annotations `{ key: "conductor", value: true }` and `{ key: "kind", value: "sequence" }`. Secondary activation records and activation records for component actions include the annotation `{ key: "causedBy", value: "sequence" }`.
+
+The memory limit annotation in the primary activation record reflects the maximum memory limit across the conductor action and the component actions.
+
+## Continuations
+
+A conductor action should return either an _error_ dictionary, i.e., a dictionary with an _error_ field, or a _continuation_, i.e., a dictionary with up to three fields `{ action, params, state }`. In essence, a continuation specifies what component action to invoke if any, as well as the parameters for this invocation, and the state to preserve until the next secondary activation of the conductor action.
+
+The execution flow in our example is the following:
+
+ 1. The _tripleAndIncrement_ action is invoked on the input dictionary `{ value: 3 }`. It returns `{ action: 'triple', params: { value: 3 }, state: { $step: 1 } }` requesting that action _triple_ be invoked on _params_ dictionary `{ value: 3 }`.
+ 2. The _triple_ action is invoked on dictionary `{ value: 3 }` returning `{ value: 9 }`.
+ 3. The _tripleAndIncrement_ action is automatically reactivated. The input dictionary for this activation is `{ value: 9, $step: 1 }` obtained by combining the result of the _triple_ action invocation with the _state_ of the prior secondary _tripleAndIncrement_ activation (see below for details). It returns `{ action: 'increment', params: { value: 9 }, state: { $step: 2 } }`.
+ 4. The _increment_ action is invoked on dictionary `{ value: 9 }` returning `{ value: 10 }`.
+ 5. The _tripleAndIncrement_ action is automatically reactivated on dictionary `{ value: 10, $step: 2 }` returning `{ params: { value: 10 } }`.
+ 6. Because the output of the last secondary _tripleAndIncrement_ activation specifies no further action to invoke, this completes the execution resulting in the recording of the primary activation. The result of the primary activation is obtained from the result of the last secondary activation by extracting the value of the _params_ field: `{ value: 10 }`.
+
+### Detailed specification
+
+If a secondary activation returns an error dictionary, the conductor action invocation ends and the result of this activation (output and status code) are those of this secondary activation.
+
+In a continuation dictionary, the _params_ field is optional and its value if present should be a dictionary. The _action_ field is optional and its value if present should be a string. The _state_ field is optional and its value if present should be a dictionary. If the value _v_ of the _params_ field is not a dictionary it is automatically boxed into dictionary `{ value: v }`. If the value _v_ of the _state_ field is not a dictionary it is automatically boxed into dictionary `{ state: v }`.
+
+If the _action_ field is defined in the output of the conductor action, the runtime attempts to convert its value (irrespective of its type) into the fully qualified name of an action and invoke this action (using the default namespace if necessary). There are four failure modes:
+
+- parsing failure,
+- resolution failure,
+- entitlement check failure,
+- internal error (invocation failure or timeout).
+
+In any of the first three failure scenarios, the conductor action invocation ends with an _application error_ status code and an error message describing the reason for the failure. In the latter, the status code is _internal error_.
+
+If there is no error, _action_ is invoked on the _params_ dictionary if specified (auto boxed if necessary) or if not on the empty dictionary.
+
+Upon completion of this invocation, the conductor action is activated again. The input dictionary for this activation is a combination of the output dictionary for the component action and the value of the _state_ field from the prior secondary conductor activation. Fields of the _state_ dictionary (auto boxed if necessary) are added to the output dictionary of the component activation, overriding values of existing fields if necessary.
+
+On the other hand, if the _action_ field is not defined in the output of the conductor action, the conductor action invocation ends. The output for the conductor action invocation is either the value of the _params_ field in the output dictionary of the last secondary activation if defined (auto boxed if necessary) or if absent the complete output dictionary.
+
+## Limits
+
+There are limits on the number of component action activations and secondary conductor activations in a conductor action invocation. These limits are assessed globally, i.e., if some components of a conductor action invocation are themselves conductor actions, the limits apply to the combined counts of activations across all the conductor action invocations.
+
+The maximum number _n_ of permitted component activations is equal to the maximum number of components in a sequence action. It is configured via the same configuration parameter. The maximum number of secondary conductor activations is _2n+1_.
+
+If either of these limits is exceeded, the conductor action invocation ends with an _application error_ status code and an error message describing the reason for the failure.
+
+
diff --git a/tests/dat/actions/conductor.js b/tests/dat/actions/conductor.js
new file mode 100644
index 0000000..30a7171
--- /dev/null
+++ b/tests/dat/actions/conductor.js
@@ -0,0 +1,15 @@
+/**
+ * Minimal conductor action.
+ */
+function main(args) {
+ // propagate errors
+ if (args.error) return args
+ // unescape params: { action, state, foo, params: { bar } } becomes { action, state, params: { foo, bar } }
+ const action = args.action
+ const state = args.state
+ const params = args.params
+ delete args.action
+ delete args.state
+ delete args.params
+ return { action, state, params: Object.assign(args, params) }
+}
diff --git a/tests/dat/actions/step.js b/tests/dat/actions/step.js
new file mode 100644
index 0000000..58c7da0
--- /dev/null
+++ b/tests/dat/actions/step.js
@@ -0,0 +1,7 @@
+/**
+ * Increment action.
+ */
+function main({ n }) {
+ if (typeof n === 'undefined') return { error: 'missing parameter'}
+ return { n: n + 1 }
+}
diff --git a/tests/src/test/scala/system/basic/WskConductorTests.scala b/tests/src/test/scala/system/basic/WskConductorTests.scala
new file mode 100644
index 0000000..2ff00ff
--- /dev/null
+++ b/tests/src/test/scala/system/basic/WskConductorTests.scala
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package system.basic
+
+import scala.concurrent.duration.DurationInt
+import scala.language.postfixOps
+
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+
+import common.ActivationResult
+import common.StreamLogging
+import common.JsHelpers
+import common.TestHelpers
+import common.TestUtils
+import common.BaseWsk
+import common.WskProps
+import common.WskTestHelpers
+
+import spray.json._
+import spray.json.DefaultJsonProtocol._
+import spray.json.JsObject
+import spray.json.pimpAny
+
+import whisk.core.entity.size.SizeInt
+import whisk.core.WhiskConfig
+import whisk.http.Messages._
+
+@RunWith(classOf[JUnitRunner])
+abstract class WskConductorTests extends TestHelpers with WskTestHelpers with JsHelpers with StreamLogging {
+
+ implicit val wskprops = WskProps()
+ val wsk: BaseWsk
+ val allowedActionDuration = 120 seconds
+
+ val testString = "this is a test"
+ val invalid = "invalid#Action"
+ val missing = "missingAction"
+
+ val whiskConfig = new WhiskConfig(Map(WhiskConfig.actionSequenceMaxLimit -> null))
+ assert(whiskConfig.isValid)
+ val limit = whiskConfig.actionSequenceLimit.toInt
+
+ behavior of "Whisk conductor actions"
+
+ it should "invoke a conductor action with no continuation" in withAssetCleaner(wskprops) { (wp, assetHelper) =>
+ val echo = "echo" // echo conductor action
+ assetHelper.withCleaner(wsk.action, echo) { (action, _) =>
+ action.create(
+ echo,
+ Some(TestUtils.getTestActionFilename("echo.js")),
+ annotations = Map("conductor" -> true.toJson))
+ }
+
+ // the conductor annotation should not affect the behavior of an action
+ // that returns a dictionary without a params or action field
+ val run = wsk.action.invoke(echo, Map("payload" -> testString.toJson, "state" -> testString.toJson))
+ withActivation(wsk.activation, run) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("payload" -> testString.toJson, "state" -> testString.toJson))
+ checkConductorLogsAndAnnotations(activation, 1) // echo
+ }
+
+ // the conductor annotation should not affect the behavior of an action that returns an error
+ val secondrun = wsk.action.invoke(echo, Map("error" -> testString.toJson))
+ withActivation(wsk.activation, secondrun) { activation =>
+ activation.response.status shouldBe "application error"
+ activation.response.result shouldBe Some(JsObject("error" -> testString.toJson))
+ checkConductorLogsAndAnnotations(activation, 1) // echo
+ }
+
+ // the controller should unwrap a wrapped result { params: result, ... } for an action with a conductor annotation
+ // discarding other fields if there is no action field
+ val thirdrun = wsk.action.invoke(
+ echo,
+ Map(
+ "params" -> JsObject("payload" -> testString.toJson),
+ "result" -> testString.toJson,
+ "state" -> testString.toJson))
+ withActivation(wsk.activation, thirdrun) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("payload" -> testString.toJson))
+ checkConductorLogsAndAnnotations(activation, 1) // echo
+ }
+ }
+
+ it should "invoke a conductor action with an invalid continuation" in withAssetCleaner(wskprops) {
+ (wp, assetHelper) =>
+ val echo = "echo" // echo conductor action
+ assetHelper.withCleaner(wsk.action, echo) { (action, _) =>
+ action.create(
+ echo,
+ Some(TestUtils.getTestActionFilename("echo.js")),
+ annotations = Map("conductor" -> true.toJson))
+ }
+
+ // an invalid action name
+ val invalidrun =
+ wsk.action.invoke(echo, Map("payload" -> testString.toJson, "action" -> invalid.toJson))
+ withActivation(wsk.activation, invalidrun) { activation =>
+ activation.response.status shouldBe "application error"
+ activation.response.result.get.fields.get("error") shouldBe Some(
+ JsString(compositionComponentInvalid(JsString(invalid))))
+ checkConductorLogsAndAnnotations(activation, 1) // echo
+ }
+
+ // an undefined action
+ val undefinedrun = wsk.action.invoke(echo, Map("payload" -> testString.toJson, "action" -> missing.toJson))
+ withActivation(wsk.activation, undefinedrun) { activation =>
+ activation.response.status shouldBe "application error"
+ activation.response.result.get.fields.get("error") shouldBe Some(
+ JsString(compositionComponentNotFound(missing)))
+ checkConductorLogsAndAnnotations(activation, 1) // echo
+ }
+ }
+
+ it should "invoke a conductor action with a continuation" in withAssetCleaner(wskprops) { (wp, assetHelper) =>
+ val conductor = "conductor" // conductor action
+ assetHelper.withCleaner(wsk.action, conductor) { (action, _) =>
+ action.create(
+ conductor,
+ Some(TestUtils.getTestActionFilename("conductor.js")),
+ annotations = Map("conductor" -> true.toJson))
+ }
+
+ val step = "step" // step action with higher memory limit than conductor to test max memory computation
+ assetHelper.withCleaner(wsk.action, step) { (action, _) =>
+ action.create(step, Some(TestUtils.getTestActionFilename("step.js")), memory = Some(257 MB))
+ }
+
+ // dynamically invoke step action
+ val run = wsk.action.invoke(conductor, Map("action" -> step.toJson, "n" -> 1.toJson))
+ withActivation(wsk.activation, run) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 2.toJson))
+ checkConductorLogsAndAnnotations(activation, 3) // conductor, step, conductor
+ }
+
+ // dynamically invoke step action with an error result
+ val errorrun = wsk.action.invoke(conductor, Map("action" -> step.toJson))
+ withActivation(wsk.activation, errorrun) { activation =>
+ activation.response.status shouldBe "application error"
+ activation.response.result shouldBe Some(JsObject("error" -> JsString("missing parameter")))
+ checkConductorLogsAndAnnotations(activation, 3) // conductor, step, conductor
+ }
+
+ // dynamically invoke step action, blocking invocation
+ val blockingrun = wsk.action.invoke(conductor, Map("action" -> step.toJson, "n" -> 1.toJson), blocking = true)
+ val activation = wsk.parseJsonString(blockingrun.stdout).convertTo[ActivationResult]
+
+ withClue(s"check failed for blocking conductor activation: $activation") {
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 2.toJson))
+ checkConductorLogsAndAnnotations(activation, 3) // conductor, step, conductor
+ }
+
+ // dynamically invoke step action, forwarding state
+ val secondrun = wsk.action.invoke(
+ conductor,
+ Map(
+ "action" -> step.toJson, // invoke step
+ "state" -> JsObject("witness" -> 42.toJson), // dummy state
+ "n" -> 1.toJson))
+ withActivation(wsk.activation, secondrun) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 2.toJson, "witness" -> 42.toJson))
+ checkConductorLogsAndAnnotations(activation, 3) // conductor, step, conductor
+ }
+
+ // dynamically invoke step action twice, forwarding state
+ val thirdrun = wsk.action.invoke(
+ conductor,
+ Map(
+ "action" -> step.toJson, // invoke step
+ "state" -> JsObject("action" -> step.toJson), // invoke step again
+ "n" -> 1.toJson))
+ withActivation(wsk.activation, thirdrun) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 3.toJson))
+ checkConductorLogsAndAnnotations(activation, 5) // conductor, step, conductor, step, conductor
+ }
+ }
+
+ it should "invoke nested conductor actions" in withAssetCleaner(wskprops) { (wp, assetHelper) =>
+ val conductor = "conductor" // conductor action
+ assetHelper.withCleaner(wsk.action, conductor) { (action, _) =>
+ action.create(
+ conductor,
+ Some(TestUtils.getTestActionFilename("conductor.js")),
+ annotations = Map("conductor" -> true.toJson))
+ }
+
+ val step = "step" // step action with lower memory limit than conductor to test max memory computation
+ assetHelper.withCleaner(wsk.action, step) { (action, _) =>
+ action.create(step, Some(TestUtils.getTestActionFilename("step.js")), memory = Some(255 MB))
+ }
+
+ // invoke nested conductor with single step
+ val run = wsk.action.invoke(
+ conductor,
+ Map(
+ "action" -> conductor.toJson, // invoke nested conductor
+ "params" -> JsObject("action" -> step.toJson), // invoke step (level 1)
+ "n" -> 1.toJson))
+ withActivation(wsk.activation, run) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 2.toJson))
+ checkConductorLogsAndAnnotations(activation, 3) // conductor, nested conductor, conductor
+ // check nested conductor invocation
+ withActivation(
+ wsk.activation,
+ activation.logs.get(1),
+ initialWait = 1 second,
+ pollPeriod = 60 seconds,
+ totalWait = allowedActionDuration) { nestedActivation =>
+ nestedActivation.response.status shouldBe "success"
+ nestedActivation.response.result shouldBe Some(JsObject("n" -> 2.toJson))
+ checkConductorLogsAndAnnotations(nestedActivation, 3) // conductor, step, conductor
+ }
+ }
+
+ // invoke nested conductor with single step, blocking invocation
+ val blockingrun = wsk.action.invoke(
+ conductor,
+ Map(
+ "action" -> conductor.toJson, // invoke nested conductor
+ "params" -> JsObject("action" -> step.toJson), // invoke step (level 1)
+ "n" -> 1.toJson),
+ blocking = true)
+ val activation = wsk.parseJsonString(blockingrun.stdout).convertTo[ActivationResult]
+
+ withClue(s"check failed for blocking conductor activation: $activation") {
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 2.toJson))
+ checkConductorLogsAndAnnotations(activation, 3) // conductor, nested conductor, conductor
+ }
+
+ // nested step followed by outer step
+ val secondrun = wsk.action.invoke(
+ conductor,
+ Map(
+ "action" -> conductor.toJson, // invoke nested conductor
+ "state" -> JsObject("action" -> step.toJson), // invoked step on return of nested conductor (level 0)
+ "params" -> JsObject("action" -> step.toJson), // invoke step (level 1)
+ "n" -> 1.toJson))
+ withActivation(wsk.activation, secondrun) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 3.toJson))
+ checkConductorLogsAndAnnotations(activation, 5)
+ }
+
+ // two levels of nesting, three steps
+ val thirdrun = wsk.action.invoke(
+ conductor,
+ Map(
+ "action" -> conductor.toJson, // invoke nested conductor
+ "state" -> JsObject("action" -> step.toJson), // invoke step on return (level 0)
+ "params" -> JsObject(
+ "action" -> conductor.toJson, // invoked nested nested conductor
+ "state" -> JsObject("action" -> step.toJson), // invoke step on return (level 1)
+ "params" -> JsObject("action" -> step.toJson)), // invoke step (level 2)
+ "n" -> 1.toJson))
+ withActivation(wsk.activation, thirdrun) { activation =>
+ activation.response.status shouldBe "success"
+ activation.response.result shouldBe Some(JsObject("n" -> 4.toJson))
+ checkConductorLogsAndAnnotations(activation, 5)
+ }
+ }
+
+ /**
+ * checks logs for the activation of a conductor action (length/size and ids)
+ * checks that the cause field for nested invocations is set properly
+ * checks duration
+ * checks memory
+ */
+ private def checkConductorLogsAndAnnotations(activation: ActivationResult, size: Int) = {
+ activation.logs shouldBe defined
+ // check that the logs are what they are supposed to be (activation ids)
+ // check that the cause field is properly set for these activations
+ activation.logs.get should have length size // the number of activations in this sequence
+ var totalTime: Long = 0
+ var maxMemory: Long = 0
+ activation.logs.get.foreach { id =>
+ withActivation(
+ wsk.activation,
+ id,
+ initialWait = 1 second,
+ pollPeriod = 60 seconds,
+ totalWait = allowedActionDuration) { componentActivation =>
+ componentActivation.cause shouldBe defined
+ componentActivation.cause.get shouldBe (activation.activationId)
+ // check causedBy
+ val causedBy = componentActivation.getAnnotationValue("causedBy")
+ causedBy shouldBe defined
+ causedBy.get shouldBe (JsString("sequence"))
+ totalTime += componentActivation.duration
+ // extract memory
+ val mem = extractMemoryAnnotation(componentActivation)
+ maxMemory = maxMemory max mem
+ }
+ }
+ // extract duration
+ activation.duration shouldBe (totalTime)
+ // extract memory
+ activation.annotations shouldBe defined
+ val memory = extractMemoryAnnotation(activation)
+ memory shouldBe (maxMemory)
+ }
+
+ private def extractMemoryAnnotation(activation: ActivationResult): Long = {
+ val limits = activation.getAnnotationValue("limits")
+ limits shouldBe defined
+ limits.get.asJsObject.getFields("memory")(0).convertTo[Long]
+ }
+}
diff --git a/tests/src/test/scala/system/basic/WskRestConductorTests.scala b/tests/src/test/scala/system/basic/WskRestConductorTests.scala
new file mode 100644
index 0000000..c43bcdb
--- /dev/null
+++ b/tests/src/test/scala/system/basic/WskRestConductorTests.scala
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package system.basic
+
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+
+import common.rest.WskRest
+
+@RunWith(classOf[JUnitRunner])
+class WskRestConductorTests extends WskConductorTests {
+ override val wsk: WskRest = new WskRest
+}
diff --git a/tests/src/test/scala/whisk/core/controller/test/ConductorsApiTests.scala b/tests/src/test/scala/whisk/core/controller/test/ConductorsApiTests.scala
new file mode 100644
index 0000000..f08d700
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/controller/test/ConductorsApiTests.scala
@@ -0,0 +1,374 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.controller.test
+
+import java.time.Instant
+
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext
+import scala.language.postfixOps
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+import akka.http.scaladsl.model.StatusCodes._
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport.sprayJsonMarshaller
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport.sprayJsonUnmarshaller
+import akka.http.scaladsl.server.Route
+import spray.json._
+import spray.json.DefaultJsonProtocol._
+import whisk.common.TransactionId
+import whisk.core.WhiskConfig
+import whisk.core.connector.ActivationMessage
+import whisk.core.controller.WhiskActionsApi
+import whisk.core.entity._
+import whisk.http.Messages._
+
+/**
+ * Tests Conductor Actions API.
+ *
+ * Unit tests of the controller service as a standalone component.
+ * These tests exercise a fresh instance of the service object in memory -- these
+ * tests do NOT communication with a whisk deployment.
+ */
+@RunWith(classOf[JUnitRunner])
+class ConductorsApiTests extends ControllerTestCommon with WhiskActionsApi {
+
+ /** Conductors API tests */
+ behavior of "Conductor"
+
+ val creds = WhiskAuthHelpers.newIdentity()
+ val namespace = EntityPath(creds.subject.asString)
+ val collectionPath = s"/${EntityPath.DEFAULT}/${collection.path}"
+
+ val alternateCreds = WhiskAuthHelpers.newIdentity()
+ val alternateNamespace = EntityPath(alternateCreds.subject.asString)
+
+ // test actions
+ val echo = MakeName.next("echo")
+ val conductor = MakeName.next("conductor")
+ val step = MakeName.next("step")
+ val missing = MakeName.next("missingAction") // undefined
+ val invalid = "invalid#Action" // invalid name
+
+ val testString = "this is a test"
+ val duration = 42
+
+ val limit = whiskConfig.actionSequenceLimit.toInt
+
+ override val loadBalancer = new FakeLoadBalancerService(whiskConfig)
+ override val activationIdFactory = new ActivationId.ActivationIdGenerator() {}
+
+ it should "invoke a conductor action with no dynamic steps" in {
+ implicit val tid = transid()
+ put(entityStore, WhiskAction(namespace, echo, jsDefault("??"), annotations = Parameters("conductor", "true")))
+
+ // a normal result
+ Post(s"$collectionPath/${echo}?blocking=true", JsObject("payload" -> testString.toJson)) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("payload" -> testString.toJson)
+ response.fields("duration") shouldBe duration.toJson
+ val annotations = response.fields("annotations").convertTo[Parameters]
+ annotations.getAs[Boolean]("conductor") shouldBe Some(true)
+ annotations.getAs[String]("kind") shouldBe Some("sequence")
+ annotations.getAs[Boolean]("topmost") shouldBe Some(true)
+ annotations.get("limits") should not be None
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 1
+ }
+
+ // an error result
+ Post(s"$collectionPath/${echo}?blocking=true", JsObject("error" -> testString.toJson)) ~> Route.seal(routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("error" -> testString.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 1
+ }
+
+ // a wrapped result { params: result } is unwrapped by the controller
+ Post(s"$collectionPath/${echo}?blocking=true", JsObject("params" -> JsObject("payload" -> testString.toJson))) ~> Route
+ .seal(routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("payload" -> testString.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 1
+ }
+
+ // an invalid action name
+ Post(s"$collectionPath/${echo}?blocking=true", JsObject("payload" -> testString.toJson, "action" -> invalid.toJson)) ~> Route
+ .seal(routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject(
+ "error" -> compositionComponentInvalid(invalid.toJson).toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 1
+ }
+
+ // an undefined action
+ Post(s"$collectionPath/${echo}?blocking=true", JsObject("payload" -> testString.toJson, "action" -> missing.toJson)) ~> Route
+ .seal(routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject(
+ "error" -> compositionComponentNotFound(missing.toString).toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 1
+ }
+ }
+
+ it should "invoke a conductor action with dynamic steps" in {
+ implicit val tid = transid()
+ put(entityStore, WhiskAction(namespace, conductor, jsDefault("??"), annotations = Parameters("conductor", "true")))
+ put(entityStore, WhiskAction(namespace, step, jsDefault("??")))
+ put(entityStore, WhiskAction(alternateNamespace, step, jsDefault("??"))) // forbidden action
+ val forbidden = s"$alternateNamespace/$step" // forbidden action name
+
+ // dynamically invoke step action
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> step.toJson, "params" -> JsObject("n" -> 1.toJson))) ~> Route.seal(routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 2.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 3
+ response.fields("duration") shouldBe (3 * duration).toJson
+ }
+
+ // dynamically invoke step action with an error result
+ Post(s"$collectionPath/${conductor}?blocking=true", JsObject("action" -> step.toJson)) ~> Route.seal(routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("error" -> "missing parameter".toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 3
+ response.fields("duration") shouldBe (3 * duration).toJson
+ }
+
+ // dynamically invoke step action, forwarding state
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> step.toJson, "state" -> JsObject("witness" -> 42.toJson), "n" -> 1.toJson)) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 2.toJson, "witness" -> 42.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 3
+ response.fields("duration") shouldBe (3 * duration).toJson
+ }
+
+ // dynamically invoke a forbidden action
+ Post(s"$collectionPath/${conductor}?blocking=true", JsObject("action" -> forbidden.toJson)) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject(
+ "error" -> compositionComponentNotAccessible(forbidden).toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 1
+ }
+
+ // dynamically invoke step action twice, forwarding state
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> step.toJson, "state" -> JsObject("action" -> step.toJson), "n" -> 1.toJson)) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 3.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 5
+ response.fields("duration") shouldBe (5 * duration).toJson
+ }
+
+ // invoke nested conductor with single step
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> conductor.toJson, "params" -> JsObject("action" -> step.toJson), "n" -> 1.toJson)) ~> Route
+ .seal(routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 2.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 3
+ response.fields("duration") shouldBe (5 * duration).toJson
+ }
+
+ // nested step followed by outer step
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject(
+ "action" -> conductor.toJson,
+ "state" -> JsObject("action" -> step.toJson),
+ "params" -> JsObject("action" -> step.toJson),
+ "n" -> 1.toJson)) ~> Route.seal(routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 3.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 5
+ response.fields("duration") shouldBe (7 * duration).toJson
+ }
+
+ // two levels of nesting, three steps
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject(
+ "action" -> conductor.toJson,
+ "state" -> JsObject("action" -> step.toJson),
+ "params" -> JsObject(
+ "action" -> conductor.toJson,
+ "state" -> JsObject("action" -> step.toJson),
+ "params" -> JsObject("action" -> step.toJson)),
+ "n" -> 1.toJson)) ~> Route.seal(routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 4.toJson)
+ response.fields("logs").convertTo[JsArray].elements.size shouldBe 5
+ response.fields("duration") shouldBe (11 * duration).toJson
+ }
+ }
+
+ it should "abort if composition is too long" in {
+ implicit val tid = transid()
+ put(entityStore, WhiskAction(namespace, conductor, jsDefault("??"), annotations = Parameters("conductor", "true")))
+ put(entityStore, WhiskAction(namespace, step, jsDefault("??")))
+
+ // stay just below limit
+ var params = Map[String, JsValue]()
+ for (i <- 1 to limit) {
+ params = Map("action" -> step.toJson, "state" -> JsObject(params))
+ }
+ Post(s"$collectionPath/${conductor}?blocking=true", JsObject(params + ("n" -> 0.toJson))) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> limit.toJson)
+ response.fields("duration") shouldBe (101 * duration).toJson
+ }
+
+ // add one extra step
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> step.toJson, "state" -> JsObject(params), "n" -> 0.toJson)) ~> Route.seal(routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("error" -> compositionIsTooLong.toJson)
+ }
+
+ // nesting a composition at the limit should be ok
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> conductor.toJson, "params" -> JsObject(params), "n" -> 0.toJson)) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> limit.toJson)
+ }
+
+ // nesting a composition beyond the limit should fail
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject(
+ "action" -> conductor.toJson,
+ "params" -> JsObject("action" -> step.toJson, "state" -> JsObject(params)),
+ "n" -> 0.toJson)) ~> Route.seal(routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("error" -> compositionIsTooLong.toJson)
+ }
+
+ // recursing at the limit should be ok
+ params = Map[String, JsValue]()
+ for (i <- 1 to limit) {
+ params = Map("action" -> conductor.toJson, "params" -> JsObject(params))
+ }
+ Post(s"$collectionPath/${conductor}?blocking=true", JsObject(params + ("n" -> 0.toJson))) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should be(OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("n" -> 0.toJson)
+ }
+
+ // recursing beyond the limit should fail
+ Post(
+ s"$collectionPath/${conductor}?blocking=true",
+ JsObject("action" -> conductor.toJson, "params" -> JsObject(params), "n" -> 0.toJson)) ~> Route.seal(
+ routes(creds)) ~> check {
+ status should not be (OK)
+ val response = responseAs[JsObject]
+ response.fields("response").asJsObject.fields("status") shouldBe "application error".toJson
+ response.fields("response").asJsObject.fields("result") shouldBe JsObject("error" -> compositionIsTooLong.toJson)
+ }
+ }
+
+ // fake load balancer to emulate a handful of actions
+ class FakeLoadBalancerService(config: WhiskConfig)(implicit ec: ExecutionContext)
+ extends DegenerateLoadBalancerService(config) {
+
+ private def respond(action: ExecutableWhiskActionMetaData, msg: ActivationMessage, result: JsObject) = {
+ val response =
+ if (result.fields.get("error") isDefined) ActivationResponse(ActivationResponse.ApplicationError, Some(result))
+ else ActivationResponse.success(Some(result))
+ val start = Instant.now
+ WhiskActivation(
+ action.namespace,
+ action.name,
+ msg.user.subject,
+ msg.activationId,
+ start,
+ end = start.plusMillis(duration),
+ response = response)
+ }
+
+ override def publish(action: ExecutableWhiskActionMetaData, msg: ActivationMessage)(
+ implicit transid: TransactionId): Future[Future[Either[ActivationId, WhiskActivation]]] =
+ msg.content map { args =>
+ Future.successful {
+ action.name match {
+ case `echo` => // echo action
+ Future(Right(respond(action, msg, args)))
+ case `conductor` => // see tests/dat/actions/conductor.js
+ val result =
+ if (args.fields.get("error") isDefined) args
+ else {
+ val action = args.fields.get("action") map { action =>
+ Map("action" -> action)
+ } getOrElse Map()
+ val state = args.fields.get("state") map { state =>
+ Map("state" -> state)
+ } getOrElse Map()
+ val wrappedParams = args.fields.getOrElse("params", JsObject()).asJsObject.fields
+ val escapedParams = args.fields - "action" - "state" - "params"
+ val params = Map("params" -> JsObject(wrappedParams ++ escapedParams))
+ JsObject(params ++ action ++ state)
+ }
+ Future(Right(respond(action, msg, result)))
+ case `step` => // see tests/dat/actions/step.js
+ val result = args.fields.get("n") map { n =>
+ JsObject("n" -> (n.convertTo[BigDecimal] + 1).toJson)
+ } getOrElse {
+ JsObject("error" -> "missing parameter".toJson)
+ }
+ Future(Right(respond(action, msg, result)))
+ case _ =>
+ Future.failed(new IllegalArgumentException("Unkown action invoked in conductor test"))
+ }
+ }
+ } getOrElse Future.failed(new IllegalArgumentException("No invocation parameters in conductor test"))
+ }
+}
--
To stop receiving notification emails like this one, please contact
rabbah@apache.org.