This is an automated email from the ASF dual-hosted git repository. csantanapr 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 6720d2e Add init and wait time to activation record as annotation. (#3053) 6720d2e is described below commit 6720d2e850bd989e20f685ca4674906463b55b71 Author: rodric rabbah <rod...@gmail.com> AuthorDate: Wed Dec 13 10:52:07 2017 -0500 Add init and wait time to activation record as annotation. (#3053) Annotate an activation record with two new properties: initTime and waitTime. The former is present for cold starts and reflects the time spent in /init. The latter is the time from the controller receiving the request to the time the invoker initiated a /init or /run for the request against a container. Is it a rough indicator of the queueing time for the request. It is only present for non-sequence activations. --- .../scala/whisk/core/containerpool/Container.scala | 1 + .../main/scala/whisk/core/entity/Parameter.scala | 8 ++- .../scala/whisk/core/entity/WhiskActivation.scala | 11 ++- .../core/controller/actions/SequenceActions.scala | 16 ++--- .../whisk/core/containerpool/ContainerProxy.scala | 57 ++++++++++----- .../scala/whisk/core/invoker/InvokerReactive.scala | 7 +- docs/annotations.md | 83 +++++++++++++++++----- .../containerpool/test/ContainerProxyTests.scala | 69 ++++++++++++++++-- 8 files changed, 198 insertions(+), 54 deletions(-) diff --git a/common/scala/src/main/scala/whisk/core/containerpool/Container.scala b/common/scala/src/main/scala/whisk/core/containerpool/Container.scala index fae0f2d..f3ebcf3 100644 --- a/common/scala/src/main/scala/whisk/core/containerpool/Container.scala +++ b/common/scala/src/main/scala/whisk/core/containerpool/Container.scala @@ -191,6 +191,7 @@ case class RunResult(interval: Interval, response: Either[ContainerConnectionErr def ok = response.right.exists(_.ok) def toBriefString = response.fold(_.toString, _.toString) } + object Interval { /** An interval starting now with zero duration. */ 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 660c6e6..7355a43 100644 --- a/common/scala/src/main/scala/whisk/core/entity/Parameter.scala +++ b/common/scala/src/main/scala/whisk/core/entity/Parameter.scala @@ -39,13 +39,14 @@ protected[core] class Parameters protected[entity] (private val params: Map[Para * * @return Size of instance as ByteSize */ - def size = + def size = { params .map { case (name, value) => name.size + value.size } .foldLeft(0 B)(_ + _) + } protected[entity] def +(p: (ParameterName, ParameterValue)) = { Option(p) map { p => @@ -60,6 +61,11 @@ protected[core] class Parameters protected[entity] (private val params: Map[Para /** Add parameters from p to existing map, overwriting existing values in case of overlap in keys. */ protected[core] def ++(p: Parameters) = new Parameters(params ++ p.params) + /** Add optional parameters from p to existing map, overwriting existing values in case of overlap in keys. */ + protected[core] def ++(p: Option[Parameters]): Parameters = { + p.map(x => new Parameters(params ++ x.params)).getOrElse(this) + } + /** Remove parameter by name. */ protected[core] def -(p: String) = { // wrap with try since parameter name may throw an exception for illegal p 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 1e04b11..95e890a 100644 --- a/common/scala/src/main/scala/whisk/core/entity/WhiskActivation.scala +++ b/common/scala/src/main/scala/whisk/core/entity/WhiskActivation.scala @@ -123,6 +123,15 @@ object WhiskActivation with WhiskEntityQueries[WhiskActivation] with DefaultJsonProtocol { + /** Some field names for annotations */ + val pathAnnotation = "path" + val kindAnnotation = "kind" + val limitsAnnotation = "limits" + val topmostAnnotation = "topmost" + val causedByAnnotation = "causedBy" + val initTimeAnnotation = "initTime" + val waitTimeAnnotation = "waitTime" + private implicit val instantSerdes = new RootJsonFormat[Instant] { def write(t: Instant) = t.toEpochMilli.toJson @@ -133,7 +142,7 @@ object WhiskActivation case JsNumber(i) => Instant.ofEpochMilli(i.bigDecimal.longValue) case _ => deserializationError("timetsamp malformed") } - } getOrElse deserializationError("timetsamp malformed 2") + } getOrElse deserializationError("timetsamp malformed") } override val collectionName = "activations" diff --git a/core/controller/src/main/scala/whisk/core/controller/actions/SequenceActions.scala b/core/controller/src/main/scala/whisk/core/controller/actions/SequenceActions.scala index 8df2033..90b2c2a 100644 --- a/core/controller/src/main/scala/whisk/core/controller/actions/SequenceActions.scala +++ b/core/controller/src/main/scala/whisk/core/controller/actions/SequenceActions.scala @@ -199,16 +199,14 @@ protected[actions] trait SequenceActions { // compute max memory val sequenceLimits = accounting.maxMemory map { maxMemoryAcrossActionsInSequence => Parameters( - "limits", + WhiskActivation.limitsAnnotation, ActionLimits(action.limits.timeout, MemoryLimit(maxMemoryAcrossActionsInSequence MB), action.limits.logs).toJson) - } getOrElse (Parameters()) + } // set causedBy if not topmost sequence val causedBy = if (!topmost) { - Parameters("causedBy", JsString("sequence")) - } else { - Parameters() - } + Some(Parameters(WhiskActivation.causedByAnnotation, JsString(Exec.SEQUENCE))) + } else None // create the whisk activation WhiskActivation( @@ -223,9 +221,9 @@ protected[actions] trait SequenceActions { logs = accounting.finalLogs, version = action.version, publish = false, - annotations = Parameters("topmost", JsBoolean(topmost)) ++ - Parameters("path", action.fullyQualifiedName(false).toString) ++ - Parameters("kind", "sequence") ++ + annotations = Parameters(WhiskActivation.topmostAnnotation, JsBoolean(topmost)) ++ + Parameters(WhiskActivation.pathAnnotation, JsString(action.fullyQualifiedName(false).asString)) ++ + Parameters(WhiskActivation.kindAnnotation, JsString(Exec.SEQUENCE)) ++ causedBy ++ sequenceLimits, duration = Some(accounting.duration)) diff --git a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala index beee654..b76a60d 100644 --- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala +++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala @@ -152,7 +152,7 @@ class ContainerProxy( // also update the feed and active ack; the container cleanup is queued // implicitly via a FailureMessage which will be processed later when the state // transitions to Running - val activation = ContainerProxy.constructWhiskActivation(job, Interval.zero, response) + val activation = ContainerProxy.constructWhiskActivation(job, None, Interval.zero, response) sendActiveAck(transid, activation, job.msg.blocking, job.msg.rootControllerIndex) storeActivation(transid, activation) } @@ -325,8 +325,8 @@ class ContainerProxy( // Only initialize iff we haven't yet warmed the container val initialize = stateData match { - case data: WarmedData => Future.successful(Interval.zero) - case _ => container.initialize(job.action.containerInitializer, actionTimeout) + case data: WarmedData => Future.successful(None) + case _ => container.initialize(job.action.containerInitializer, actionTimeout).map(Some(_)) } val activation: Future[WhiskActivation] = initialize @@ -344,19 +344,21 @@ class ContainerProxy( container.run(parameters, environment, actionTimeout)(job.msg.transid).map { case (runInterval, response) => - val initRunInterval = - Interval(runInterval.start.minusMillis(initInterval.duration.toMillis), runInterval.end) - ContainerProxy.constructWhiskActivation(job, initRunInterval, response) + val initRunInterval = initInterval + .map(i => Interval(runInterval.start.minusMillis(i.duration.toMillis), runInterval.end)) + .getOrElse(runInterval) + ContainerProxy.constructWhiskActivation(job, initInterval, initRunInterval, response) } } .recover { case InitializationError(interval, response) => - ContainerProxy.constructWhiskActivation(job, interval, response) + ContainerProxy.constructWhiskActivation(job, Some(interval), interval, response) case t => // Actually, this should never happen - but we want to make sure to not miss a problem logging.error(this, s"caught unexpected error while running activation: ${t}") ContainerProxy.constructWhiskActivation( job, + None, Interval.zero, ActivationResponse.whiskError(Messages.abnormalRun)) } @@ -426,8 +428,31 @@ object ContainerProxy { * @param response the response to return to the user * @return a WhiskActivation to be sent to the user */ - def constructWhiskActivation(job: Run, interval: Interval, response: ActivationResponse) = { - val causedBy = if (job.msg.causedBySequence) Parameters("causedBy", "sequence".toJson) else Parameters() + def constructWhiskActivation(job: Run, + initInterval: Option[Interval], + totalInterval: Interval, + response: ActivationResponse) = { + val causedBy = Some { + if (job.msg.causedBySequence) { + Parameters(WhiskActivation.causedByAnnotation, JsString(Exec.SEQUENCE)) + } else { + // emit the internal system hold time as the 'wait' time, but only for non-sequence + // actions, since the transid start time for a sequence does not correspond + // with a specific component of the activation but the entire sequence; + // it will require some work to generate a new transaction id for a sequence + // component - however, because the trace of activations is recorded in the parent + // sequence, a client can determine the queue time for sequences that way + val end = initInterval.map(_.start).getOrElse(totalInterval.start) + Parameters( + WhiskActivation.waitTimeAnnotation, + Interval(job.msg.transid.meta.start, end).duration.toMillis.toJson) + } + } + + val initTime = { + initInterval.map(initTime => Parameters(WhiskActivation.initTimeAnnotation, initTime.duration.toMillis.toJson)) + } + WhiskActivation( activationId = job.msg.activationId, namespace = job.msg.activationNamespace, @@ -435,15 +460,15 @@ object ContainerProxy { cause = job.msg.cause, name = job.action.name, version = job.action.version, - start = interval.start, - end = interval.end, - duration = Some(interval.duration.toMillis), + start = totalInterval.start, + end = totalInterval.end, + duration = Some(totalInterval.duration.toMillis), response = response, annotations = { - Parameters("limits", job.action.limits.toJson) ++ - Parameters("path", job.action.fullyQualifiedName(false).toString.toJson) ++ - Parameters("kind", job.action.exec.kind.toJson) ++ - causedBy + Parameters(WhiskActivation.limitsAnnotation, job.action.limits.toJson) ++ + Parameters(WhiskActivation.pathAnnotation, JsString(job.action.fullyQualifiedName(false).asString)) ++ + Parameters(WhiskActivation.kindAnnotation, JsString(job.action.exec.kind)) ++ + causedBy ++ initTime }) } } diff --git a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala index 6a4adca..5800521 100644 --- a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala +++ b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala @@ -32,7 +32,6 @@ import akka.actor.ActorSystem import akka.actor.Props import akka.stream.ActorMaterializer import spray.json._ -import spray.json.DefaultJsonProtocol._ import whisk.common.Logging import whisk.common.LoggingMarkers import whisk.common.TransactionId @@ -206,7 +205,9 @@ class InvokerReactive(config: WhiskConfig, instance: InstanceId, producer: Messa case _ => ActivationResponse.whiskError(Messages.actionMismatchWhileInvoking) } val now = Instant.now - val causedBy = if (msg.causedBySequence) Parameters("causedBy", "sequence".toJson) else Parameters() + val causedBy = if (msg.causedBySequence) { + Some(Parameters(WhiskActivation.causedByAnnotation, JsString(Exec.SEQUENCE))) + } else None val activation = WhiskActivation( activationId = msg.activationId, namespace = msg.activationNamespace, @@ -219,7 +220,7 @@ class InvokerReactive(config: WhiskConfig, instance: InstanceId, producer: Messa duration = Some(0), response = response, annotations = { - Parameters("path", msg.action.toString.toJson) ++ causedBy + Parameters(WhiskActivation.pathAnnotation, JsString(msg.action.asString)) ++ causedBy }) activationFeed ! MessageFeed.Processed diff --git a/docs/annotations.md b/docs/annotations.md index 588d1f2..6b0ab8d 100644 --- a/docs/annotations.md +++ b/docs/annotations.md @@ -18,24 +18,24 @@ wsk action create echo echo.js \ The annotations we have used for describing packages are: -1. `description`: a pithy description of the package -2. `parameters`: an array describing parameters that are scoped to the package (described further below) +* `description`: a pithy description of the package +* `parameters`: an array describing parameters that are scoped to the package (described further below) Similarly, for actions: -1. `description`: a pithy description of the action -2. `parameters`: an array describing actions that are required to execute the action -3. `sampleInput`: an example showing the input schema with typical values -4. `sampleOutput`: an example showing the output schema, usually for the `sampleInput` +* `description`: a pithy description of the action +* `parameters`: an array describing actions that are required to execute the action +* `sampleInput`: an example showing the input schema with typical values +* `sampleOutput`: an example showing the output schema, usually for the `sampleInput` The annotations we have used for describing parameters include: -1. `name`: the name of the parameter -2. `description`: a pithy description of the parameter -3. `doclink`: a link to further documentation for parameter (useful for OAuth tokens for example) -4. `required`: true for required parameters and false for optional ones -5. `bindTime`: true if the parameter should be specified when a package is bound -6. `type`: the type of the parameter, one of `password`, `array` (but may be used more broadly) +* `name`: the name of the parameter +* `description`: a pithy description of the parameter +* `doclink`: a link to further documentation for parameter (useful for OAuth tokens for example) +* `required`: true for required parameters and false for optional ones +* `bindTime`: true if the parameter should be specified when a package is bound +* `type`: the type of the parameter, one of `password`, `array` (but may be used more broadly) The annotations are _not_ checked. So while it is conceivable to use the annotations to infer if a composition of two actions into a sequence is legal, for example, the system does not yet do that. @@ -44,8 +44,57 @@ The annotations are _not_ checked. So while it is conceivable to use the annotat Web actions are enabled with explicit annotations which decorate individual actions. The annotations only apply to the [web actions](webactions.md) API, and must be present and explicitly set to `true` to have an affect. The annotations have no meaning otherwise in the system. The annotations are: -1. `web-export`: Makes its corresponding action accessible to REST calls _without_ authentication. We call these [_web actions_](webactions.md) because they allow one to use OpenWhisk actions from a browser for example. It is important to note that the _owner_ of the web action incurs the cost of running them in the system (i.e., the _owner_ of the action also owns the activations record). The rest of the annotations described below have no effect on the action unless this annotation is [...] -2. `final`: Makes all of the action parameters that are already defined immutable. A parameter of an action carrying the annotation may not be overridden by invoke-time parameters once the parameter has a value defined through its enclosing package or the action definition. -3. `raw-http`: When set, the HTTP request query and body parameters are passed to the action as reserved properties. -4. `web-custom-options`: When set, this annotation enables a web action to respond to OPTIONS requests with customized headers, otherwise a [default CORS response](webactions.md#options-requests) applies. -5. `require-whisk-auth`: This annotation protects the web action so that it is only accessible to an authenticated subject. It is important to note that the _owner_ of the web action will still incur the cost of running them in the system (i.e., the _owner_ of the action also owns the activations record). +* `web-export`: Makes its corresponding action accessible to REST calls _without_ authentication. We call these [_web actions_](webactions.md) because they allow one to use OpenWhisk actions from a browser for example. It is important to note that the _owner_ of the web action incurs the cost of running them in the system (i.e., the _owner_ of the action also owns the activations record). The rest of the annotations described below have no effect on the action unless this annotation is a [...] +* `final`: Makes all of the action parameters that are already defined immutable. A parameter of an action carrying the annotation may not be overridden by invoke-time parameters once the parameter has a value defined through its enclosing package or the action definition. +* `raw-http`: When set, the HTTP request query and body parameters are passed to the action as reserved properties. +* `web-custom-options`: When set, this annotation enables a web action to respond to OPTIONS requests with customized headers, otherwise a [default CORS response](webactions.md#options-requests) applies. +* `require-whisk-auth`: This annotation protects the web action so that it is only accessible to an authenticated subject. It is important to note that the _owner_ of the web action will still incur the cost of running them in the system (i.e., the _owner_ of the action also owns the activations record). + +# Annotations specific to activations + +The system decorates activation records with annotations as well. They are: + +* `path`: the fully qualified path name of the action that generated the activation. Note that if this activation was the result of an action in a package binding, the path refers to the parent package. +* `kind`: the kind of action executed, and one of the support OpenWhisk runtime kinds. +* `limits`: the time, memory and log limits that this activation were subject to. + +Additionally for sequence related activations, the system will generate the following annotations: + +* `topmost`: this is only present for an outermost sequence action. +* `causedBy`: this is only present for actions that are contained in a sequence. + +Lastly, and in order to provide you with some performance transparency, activations also record: + +* `wait`: the time spent waiting in the internal OpenWhisk system. This is roughly the time spent between the controller receiving the activation request and when the invoker provisioned a container for the action. This value is currently only present for non-sequence related activations. For sequences, this information can be derived from the `topmost` sequence activation record. +* `init`: the time spent initializing the function. If this value is present, the action required initialization and represents a cold start. A warm activation will skip initialization, and in this case, the annotation is not generated. + +An example of these annotations as they would appear in an activation record is shown below. + +```javascript +"annotations": [ + { + "key": "path", + "value": "guest/echo" + }, + { + "key": "wait", + "value": 66 + }, + { + "key": "kind", + "value": "nodejs:6" + }, + { + "key": "init", + "value": 50 + }, + { + "key": "limits", + "value": { + "logs": 10, + "memory": 256, + "timeout": 60000 + } + } +] +``` diff --git a/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala b/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala index 6182d62..948acf6 100644 --- a/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala +++ b/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala @@ -78,8 +78,21 @@ class ContainerProxyTests val invocationNamespace = EntityName("invocationSpace") val action = ExecutableWhiskAction(EntityPath("actionSpace"), EntityName("actionName"), exec) + // create a transaction id to set the start time and control queue time + val messageTransId = TransactionId(BigDecimal(TransactionId.testing.meta.id)) + + val initInterval = { + val now = messageTransId.meta.start.plusMillis(50) // this is the queue time for cold start + Interval(now, now.plusMillis(100)) + } + + val runInterval = { + val now = initInterval.end.plusMillis(75) // delay between init and run + Interval(now, now.plusMillis(200)) + } + val message = ActivationMessage( - TransactionId.testing, + messageTransId, action.fullyQualifiedName(true), action.rev, Identity(Subject(), invocationNamespace, AuthKey(), Set()), @@ -247,6 +260,25 @@ class ContainerProxyTests container.suspendCount shouldBe 0 acker.calls should have size 2 store.calls should have size 2 + + val initRunActivation = acker.calls(0)._2 + initRunActivation.duration shouldBe Some((initInterval.duration + runInterval.duration).toMillis) + initRunActivation.annotations + .get(WhiskActivation.initTimeAnnotation) + .get + .convertTo[Int] shouldBe initInterval.duration.toMillis + initRunActivation.annotations + .get(WhiskActivation.waitTimeAnnotation) + .get + .convertTo[Int] shouldBe + Interval(message.transid.meta.start, initInterval.start).duration.toMillis + + val runOnlyActivation = acker.calls(1)._2 + runOnlyActivation.duration shouldBe Some(runInterval.duration.toMillis) + runOnlyActivation.annotations.get(WhiskActivation.initTimeAnnotation) shouldBe empty + runOnlyActivation.annotations.get(WhiskActivation.waitTimeAnnotation).get.convertTo[Int] shouldBe { + Interval(message.transid.meta.start, runInterval.start).duration.toMillis + } } } @@ -276,6 +308,14 @@ class ContainerProxyTests container.resumeCount shouldBe 1 acker.calls should have size 2 store.calls should have size 2 + acker + .calls(0) + ._2 + .annotations + .get(WhiskActivation.initTimeAnnotation) + .get + .convertTo[Int] shouldBe initInterval.duration.toMillis + acker.calls(1)._2.annotations.get(WhiskActivation.initTimeAnnotation) shouldBe empty } } @@ -298,6 +338,13 @@ class ContainerProxyTests collector.calls should have size 1 acker.calls should have size 1 store.calls should have size 1 + acker + .calls(0) + ._2 + .annotations + .get(WhiskActivation.initTimeAnnotation) + .get + .convertTo[Int] shouldBe initInterval.duration.toMillis } } @@ -325,7 +372,9 @@ class ContainerProxyTests collector.calls should have size 0 // gather no logs container.destroyCount shouldBe 0 // no destroying possible as no container could be obtained acker.calls should have size 1 - acker.calls(0)._2.response should be a 'whiskError + val activation = acker.calls(0)._2 + activation.response should be a 'whiskError + activation.annotations.get(WhiskActivation.initTimeAnnotation) shouldBe empty store.calls should have size 1 } } @@ -335,7 +384,7 @@ class ContainerProxyTests override def initialize(initializer: JsObject, timeout: FiniteDuration)(implicit transid: TransactionId): Future[Interval] = { initializeCount += 1 - Future.failed(InitializationError(Interval.zero, ActivationResponse.applicationError("boom"))) + Future.failed(InitializationError(initInterval, ActivationResponse.applicationError("boom"))) } } val factory = createFactory(Future.successful(container)) @@ -357,7 +406,13 @@ class ContainerProxyTests container.runCount shouldBe 0 // should not run the action collector.calls should have size 1 container.destroyCount shouldBe 1 - acker.calls(0)._2.response shouldBe ActivationResponse.applicationError("boom") + val activation = acker.calls(0)._2 + activation.response shouldBe ActivationResponse.applicationError("boom") + activation.annotations + .get(WhiskActivation.initTimeAnnotation) + .get + .convertTo[Int] shouldBe initInterval.duration.toMillis + store.calls should have size 1 } } @@ -367,7 +422,7 @@ class ContainerProxyTests override def run(parameters: JsObject, environment: JsObject, timeout: FiniteDuration)( implicit transid: TransactionId): Future[(Interval, ActivationResponse)] = { runCount += 1 - Future.successful((Interval.zero, ActivationResponse.applicationError("boom"))) + Future.successful((initInterval, ActivationResponse.applicationError("boom"))) } } val factory = createFactory(Future.successful(container)) @@ -646,7 +701,7 @@ class ContainerProxyTests initializeCount += 1 initializer shouldBe action.containerInitializer timeout shouldBe action.limits.timeout.duration - Future.successful(Interval.zero) + Future.successful(initInterval) } override def run(parameters: JsObject, environment: JsObject, timeout: FiniteDuration)( implicit transid: TransactionId): Future[(Interval, ActivationResponse)] = { @@ -662,7 +717,7 @@ class ContainerProxyTests // a freshly computed deadline, as they get computed slightly after each other deadline should (be <= maxDeadline and be >= Instant.now) - Future.successful((Interval.zero, ActivationResponse.success())) + Future.successful((runInterval, ActivationResponse.success())) } def logs(limit: ByteSize, waitForSentinel: Boolean)(implicit transid: TransactionId): Source[ByteString, Any] = ??? } -- To stop receiving notification emails like this one, please contact ['"commits@openwhisk.apache.org" <commits@openwhisk.apache.org>'].