You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openwhisk.apache.org by du...@apache.org on 2019/02/28 00:39:17 UTC
[incubator-openwhisk] branch master updated: Avoid that actions do
not stop after action timeout when logging heavily (#4299)
This is an automated email from the ASF dual-hosted git repository.
dubeejw 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 33d70bf Avoid that actions do not stop after action timeout when logging heavily (#4299)
33d70bf is described below
commit 33d70bfbb368e8c263e9f7a6fccfd03cf6655d58
Author: Steffen Rost <sr...@de.ibm.com>
AuthorDate: Thu Feb 28 00:39:10 2019 +0000
Avoid that actions do not stop after action timeout when logging heavily (#4299)
* wait for sentinel only in case no timeout occurred
* introduce timeout annotation
* nodejs:6 logging timeout action w/o busy loop
* do conversion from FiniteDuration to Duration
---
.../logging/DockerToActivationFileLogStore.scala | 5 +--
.../logging/DockerToActivationLogStore.scala | 39 ++++++++++++++++++++--
.../openwhisk/core/entity/WhiskActivation.scala | 17 ++++++----
.../core/containerpool/ContainerProxy.scala | 19 +++++++++--
tests/dat/actions/loggingTimeout.js | 36 ++++++++++++++++++++
tests/src/test/scala/common/TimingHelpers.scala | 6 ++--
.../openwhisk/core/limits/ActionLimitsTests.scala | 36 +++++++++++++++++++-
7 files changed, 140 insertions(+), 18 deletions(-)
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationFileLogStore.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationFileLogStore.scala
index 868b9cf..2c9c0a9 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationFileLogStore.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationFileLogStore.scala
@@ -123,7 +123,8 @@ class DockerToActivationFileLogStore(system: ActorSystem, destinationDirectory:
container: Container,
action: ExecutableWhiskAction): Future[ActivationLogs] = {
- val logs = container.logs(action.limits.logs.asMegaBytes, action.exec.sentinelledLogs)(transid)
+ val isTimedoutActivation = activation.isTimedoutActivation
+ val logs = logStream(transid, container, action, isTimedoutActivation)
// Adding the userId field to every written record, so any background process can properly correlate.
val userIdField = Map("namespaceId" -> user.namespace.uuid.toJson)
@@ -150,7 +151,7 @@ class DockerToActivationFileLogStore(system: ActorSystem, destinationDirectory:
logs.runWith(combined)._1.flatMap { seq =>
val possibleErrors = Set(Messages.logFailure, Messages.truncateLogs(action.limits.logs.asMegaBytes))
- val errored = seq.lastOption.exists(last => possibleErrors.exists(last.contains))
+ val errored = isTimedoutActivation || seq.lastOption.exists(last => possibleErrors.exists(last.contains))
val logs = ActivationLogs(seq.toVector)
if (!errored) {
Future.successful(logs)
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationLogStore.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationLogStore.scala
index 7fa1ac6..4173a95 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationLogStore.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationLogStore.scala
@@ -17,11 +17,13 @@
package org.apache.openwhisk.core.containerpool.logging
+import java.time.Instant
import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Flow
+import akka.stream.scaladsl.Source
import akka.util.ByteString
import org.apache.openwhisk.common.TransactionId
@@ -70,19 +72,50 @@ class DockerToActivationLogStore(system: ActorSystem) extends LogStore {
override def fetchLogs(activation: WhiskActivation, context: UserContext): Future[ActivationLogs] =
Future.successful(activation.logs)
+ /**
+ * Obtains the container's stdout and stderr output.
+ *
+ * In case of a timed out activation do not wait for a sentinel to appear but instead
+ * collect the log as is and add a message to the log that data might be missing
+ *
+ * @param transid transaction id
+ * @param container container to obtain the log from
+ * @param action action that defines the log limit
+ * @param isTimedoutActivation is activation timed out
+ *
+ * @return a vector of Strings with log lines in our own JSON format
+ */
+ protected def logStream(transid: TransactionId,
+ container: Container,
+ action: ExecutableWhiskAction,
+ isTimedoutActivation: Boolean): Source[ByteString, Any] = {
+
+ // wait for a sentinel only if no container (developer) error occurred to avoid
+ // that log collection continues if the action code still logs after timeout
+ val sentinel = action.exec.sentinelledLogs && !isTimedoutActivation
+ val logs = container.logs(action.limits.logs.asMegaBytes, sentinel)(transid)
+ val logsWithPossibleError = if (isTimedoutActivation) {
+ logs.concat(
+ Source.single(ByteString(LogLine(Instant.now.toString, "stderr", Messages.logFailure).toJson.compactPrint)))
+ } else logs
+ logsWithPossibleError
+ }
+
override def collectLogs(transid: TransactionId,
user: Identity,
activation: WhiskActivation,
container: Container,
action: ExecutableWhiskAction): Future[ActivationLogs] = {
- container
- .logs(action.limits.logs.asMegaBytes, action.exec.sentinelledLogs)(transid)
+ val isTimedoutActivation = activation.isTimedoutActivation
+ val logs = logStream(transid, container, action, isTimedoutActivation)
+
+ logs
.via(DockerToActivationLogStore.toFormattedString)
.runWith(Sink.seq)
.flatMap { seq =>
val possibleErrors = Set(Messages.logFailure, Messages.truncateLogs(action.limits.logs.asMegaBytes))
- val errored = seq.lastOption.exists(last => possibleErrors.exists(last.contains))
+ val errored = isTimedoutActivation || seq.lastOption.exists(last => possibleErrors.exists(last.contains))
val logs = ActivationLogs(seq.toVector)
if (!errored) {
Future.successful(logs)
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/entity/WhiskActivation.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/entity/WhiskActivation.scala
index 4222e51..5664c42 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/entity/WhiskActivation.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/core/entity/WhiskActivation.scala
@@ -111,19 +111,23 @@ case class WhiskActivation(namespace: EntityPath,
}
}
- def metadata = {
+ def metadata =
copy(response = response.withoutResult, annotations = Parameters(), logs = ActivationLogs())
.revision[WhiskActivation](rev)
- }
- def withoutResult = {
+
+ def withoutResult =
copy(response = response.withoutResult)
.revision[WhiskActivation](rev)
- }
- def withoutLogsOrResult = {
+
+ def withoutLogsOrResult =
copy(response = response.withoutResult, logs = ActivationLogs()).revision[WhiskActivation](rev)
- }
+
def withoutLogs = copy(logs = ActivationLogs()).revision[WhiskActivation](rev)
+
def withLogs(logs: ActivationLogs) = copy(logs = logs).revision[WhiskActivation](rev)
+
+ def isTimedoutActivation = annotations.getAs[Boolean](WhiskActivation.timeoutAnnotation).getOrElse(false)
+
}
object WhiskActivation
@@ -140,6 +144,7 @@ object WhiskActivation
val initTimeAnnotation = "initTime"
val waitTimeAnnotation = "waitTime"
val conductorAnnotation = "conductor"
+ val timeoutAnnotation = "timeout"
/** Some field names for compositions */
val actionField = "action"
diff --git a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerProxy.scala b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerProxy.scala
index 7f45cc2..7419bf3 100644
--- a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerProxy.scala
+++ b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerProxy.scala
@@ -285,7 +285,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, None, Interval.zero, response)
+ val activation = ContainerProxy.constructWhiskActivation(job, None, Interval.zero, false, response)
sendActiveAck(
transid,
activation,
@@ -581,12 +581,22 @@ class ContainerProxy(
val initRunInterval = initInterval
.map(i => Interval(runInterval.start.minusMillis(i.duration.toMillis), runInterval.end))
.getOrElse(runInterval)
- ContainerProxy.constructWhiskActivation(job, initInterval, initRunInterval, response)
+ ContainerProxy.constructWhiskActivation(
+ job,
+ initInterval,
+ initRunInterval,
+ runInterval.duration >= actionTimeout,
+ response)
}
}
.recover {
case InitializationError(interval, response) =>
- ContainerProxy.constructWhiskActivation(job, Some(interval), interval, response)
+ ContainerProxy.constructWhiskActivation(
+ job,
+ Some(interval),
+ interval,
+ interval.duration >= actionTimeout,
+ 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}")
@@ -594,6 +604,7 @@ class ContainerProxy(
job,
None,
Interval.zero,
+ false,
ActivationResponse.whiskError(Messages.abnormalRun))
}
@@ -708,6 +719,7 @@ object ContainerProxy {
def constructWhiskActivation(job: Run,
initInterval: Option[Interval],
totalInterval: Interval,
+ isTimeout: Boolean,
response: ActivationResponse) = {
val causedBy = Some {
if (job.msg.causedBySequence) {
@@ -745,6 +757,7 @@ object ContainerProxy {
Parameters(WhiskActivation.limitsAnnotation, job.action.limits.toJson) ++
Parameters(WhiskActivation.pathAnnotation, JsString(job.action.fullyQualifiedName(false).asString)) ++
Parameters(WhiskActivation.kindAnnotation, JsString(job.action.exec.kind)) ++
+ Parameters(WhiskActivation.timeoutAnnotation, JsBoolean(isTimeout)) ++
causedBy ++ initTime
})
}
diff --git a/tests/dat/actions/loggingTimeout.js b/tests/dat/actions/loggingTimeout.js
new file mode 100644
index 0000000..a386de0
--- /dev/null
+++ b/tests/dat/actions/loggingTimeout.js
@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more contributor
+// license agreements; and to You under the Apache License, Version 2.0.
+
+// This action prints log lines for a defined duration.
+// The output is throttled by a defined delay between two log lines
+// in order to keep the log size small and to stay within the log size limit.
+
+function getArg(value, defaultValue) {
+ return value ? value : defaultValue;
+}
+
+// input: { duration: <duration in millis>, delay: <delay in millis> }, e.g.
+// main( { delay: 100, duration: 10000 } );
+function main(args) {
+
+ durationMillis = getArg(args.duration, 120000);
+ delayMillis = getArg(args.delay, 100);
+
+ logLines = 0;
+ startMillis = new Date();
+
+ timeout = setInterval(function() {
+ console.log(`[${ ++logLines }] The quick brown fox jumps over the lazy dog.`);
+ }, delayMillis);
+
+ return new Promise(function(resolve, reject) {
+ setTimeout(function() {
+ clearInterval(timeout);
+ message = `hello, I'm back after ${new Date() - startMillis} ms and printed ${logLines} log lines`
+ console.log(message)
+ resolve({ message: message });
+ }, durationMillis);
+ });
+
+}
+
diff --git a/tests/src/test/scala/common/TimingHelpers.scala b/tests/src/test/scala/common/TimingHelpers.scala
index d739222..8d3b2b8 100644
--- a/tests/src/test/scala/common/TimingHelpers.scala
+++ b/tests/src/test/scala/common/TimingHelpers.scala
@@ -21,10 +21,10 @@ import java.time.Instant
import scala.concurrent.duration._
trait TimingHelpers {
- def between(start: Instant, end: Instant): Duration =
- Duration.fromNanos(java.time.Duration.between(start, end).toNanos)
+ def between(start: Instant, end: Instant): FiniteDuration =
+ FiniteDuration(java.time.Duration.between(start, end).toNanos, NANOSECONDS)
- def durationOf[A](block: => A): (Duration, A) = {
+ def durationOf[A](block: => A): (FiniteDuration, A) = {
val start = Instant.now
val value = block
val end = Instant.now
diff --git a/tests/src/test/scala/org/apache/openwhisk/core/limits/ActionLimitsTests.scala b/tests/src/test/scala/org/apache/openwhisk/core/limits/ActionLimitsTests.scala
index 23523c0..c70a8b1 100644
--- a/tests/src/test/scala/org/apache/openwhisk/core/limits/ActionLimitsTests.scala
+++ b/tests/src/test/scala/org/apache/openwhisk/core/limits/ActionLimitsTests.scala
@@ -22,6 +22,7 @@ import akka.http.scaladsl.model.StatusCodes.BadGateway
import java.io.File
import java.io.PrintWriter
import java.time.Instant
+
import scala.concurrent.duration.{Duration, DurationInt}
import scala.language.postfixOps
import org.junit.runner.RunWith
@@ -30,6 +31,7 @@ import common.ActivationResult
import common.TestHelpers
import common.TestUtils
import common.TestUtils.{BAD_REQUEST, DONTCARE_EXIT, SUCCESS_EXIT}
+import common.TimingHelpers
import common.WhiskProperties
import common.rest.WskRestOperations
import common.WskProps
@@ -51,7 +53,7 @@ import org.apache.openwhisk.core.entity.size._
import org.apache.openwhisk.http.Messages
@RunWith(classOf[JUnitRunner])
-class ActionLimitsTests extends TestHelpers with WskTestHelpers with WskActorSystem {
+class ActionLimitsTests extends TestHelpers with WskTestHelpers with WskActorSystem with TimingHelpers {
implicit val wskprops = WskProps()
val wsk = new WskRestOperations
@@ -463,4 +465,36 @@ class ActionLimitsTests extends TestHelpers with WskTestHelpers with WskActorSys
_.response.result.get.fields("error") shouldBe Messages.memoryExhausted.toJson
}
}
+
+ /**
+ * Test that a heavy logging action is interrupted within its timeout limits.
+ */
+ it should "interrupt the heavy logging action within its time limits" in withAssetCleaner(wskprops) {
+ (wp, assetHelper) =>
+ val name = s"NodeJsTestLoggingActionCausingTimeout-${System.currentTimeMillis()}"
+ assetHelper.withCleaner(wsk.action, name, confirmDelete = true) { (action, _) =>
+ action.create(
+ name,
+ Some(TestUtils.getTestActionFilename("loggingTimeout.js")),
+ timeout = Some(allowedActionDuration))
+ }
+ val duration = allowedActionDuration + 3.minutes
+ val checkDuration = allowedActionDuration + 1.minutes
+ val run =
+ wsk.action.invoke(name, Map("durationMillis" -> duration.toMillis.toJson, "delayMillis" -> 100.toJson))
+ withActivation(wsk.activation, run) { result =>
+ withClue("Activation result not as expected:") {
+ result.response.status shouldBe ActivationResponse.messageForCode(ActivationResponse.DeveloperError)
+ result.response.result.get
+ .fields("error") shouldBe Messages.timedoutActivation(allowedActionDuration, init = false).toJson
+ val logs = result.logs.get
+ logs.last should include(Messages.logFailure)
+
+ val parseLogTime = (line: String) => Instant.parse(line.split(' ').head)
+ val startTime = parseLogTime(logs.head)
+ val endTime = parseLogTime(logs.last)
+ between(startTime, endTime).toMillis should be < checkDuration.toMillis
+ }
+ }
+ }
}