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 b3a57dd Provide Artifact with File Storage Activation Store (#3991)
b3a57dd is described below
commit b3a57ddf7409cd71ebb84b85143f0920995132f2
Author: James Dubee <jwdu...@us.ibm.com>
AuthorDate: Thu Jan 24 13:53:53 2019 -0500
Provide Artifact with File Storage Activation Store (#3991)
---
.../org/apache/openwhisk/core/WhiskConfig.scala | 4 +
.../core/database/ActivationFileStorage.scala | 118 ++++++++++++++++
.../ArtifactWithFileStorageActivationStore.scala | 61 ++++++++
.../openwhisk/core/entity/WhiskActivation.scala | 14 +-
.../apache/openwhisk/core/controller/Actions.scala | 2 +-
.../openwhisk/core/controller/Activations.scala | 4 +-
.../core/controller/test/ActionsApiTests.scala | 8 +-
.../core/controller/test/ActivationsApiTests.scala | 12 +-
.../controller/test/ControllerTestCommon.scala | 8 +-
...tifactWithFileStorageActivationStoreTests.scala | 157 +++++++++++++++++++++
10 files changed, 370 insertions(+), 18 deletions(-)
diff --git
a/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
b/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
index 8061b2a..32f5a90 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
@@ -243,4 +243,8 @@ object ConfigKeys {
val controller = s"whisk.controller"
val controllerActivation = s"$controller.activation"
+
+ val activationStore = "whisk.activation-store"
+ val activationStoreWithFileStorage = s"$activationStore.with-file-storage"
+
}
diff --git
a/common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationFileStorage.scala
b/common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationFileStorage.scala
new file mode 100644
index 0000000..7fd7741
--- /dev/null
+++
b/common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationFileStorage.scala
@@ -0,0 +1,118 @@
+/*
+ * 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 org.apache.openwhisk.core.database
+
+import java.nio.file.attribute.PosixFilePermission._
+import java.nio.file.{Files, Path}
+import java.time.Instant
+import java.util.EnumSet
+
+import akka.stream.ActorMaterializer
+import akka.stream.alpakka.file.scaladsl.LogRotatorSink
+import akka.stream.scaladsl.{Flow, MergeHub, RestartSink, Sink, Source}
+import akka.util.ByteString
+import org.apache.openwhisk.common.Logging
+import
org.apache.openwhisk.core.containerpool.logging.ElasticSearchJsonProtocol._
+import org.apache.openwhisk.core.entity._
+import org.apache.openwhisk.core.entity.size._
+import spray.json._
+
+import scala.concurrent.duration._
+
+class ActivationFileStorage(logFilePrefix: String,
+ logPath: Path,
+ actorMaterializer: ActorMaterializer,
+ logging: Logging) {
+
+ implicit val materializer = actorMaterializer
+
+ private var logFile = logPath
+ private val bufferSize = 100.MB
+ private val perms = EnumSet.of(OWNER_READ, OWNER_WRITE, GROUP_READ,
GROUP_WRITE, OTHERS_READ, OTHERS_WRITE)
+ private val writeToFile: Sink[ByteString, _] = MergeHub
+ .source[ByteString]
+ .batchWeighted(bufferSize.toBytes, _.length, identity)(_ ++ _)
+ .to(RestartSink.withBackoff(minBackoff = 1.seconds, maxBackoff =
60.seconds, randomFactor = 0.2) { () =>
+ LogRotatorSink(() => {
+ val maxSize = bufferSize.toBytes
+ var bytesRead = maxSize
+ element =>
+ {
+ val size = element.size
+
+ if (bytesRead + size > maxSize) {
+ logFile =
logPath.resolve(s"$logFilePrefix-${Instant.now.toEpochMilli}.log")
+
+ logging.info(this, s"Rotating log file to '$logFile'")
+ createLogFile(logFile)
+ bytesRead = size
+ Some(logFile)
+ } else {
+ bytesRead += size
+ None
+ }
+ }
+ })
+ })
+ .run()
+
+ private def createLogFile(path: Path) =
+ try {
+ Files.createFile(path)
+ Files.setPosixFilePermissions(path, perms)
+ } catch {
+ case t: Throwable =>
+ logging.error(this, s"Couldn't create user log file '$t'")
+ throw t
+ }
+
+ private def transcribeLogs(activation: WhiskActivation, additionalFields:
Map[String, JsValue]) =
+ activation.logs.logs.map { log =>
+ val line = JsObject(
+ Map("type" -> "user_log".toJson) ++ Map("message" -> log.toJson) ++
Map(
+ "activationId" -> activation.activationId.toJson) ++
additionalFields)
+
+ ByteString(s"${line.compactPrint}\n")
+ }
+
+ private def transcribeActivation(activation: WhiskActivation,
additionalFields: Map[String, JsValue]) = {
+ val transactionType = Map("type" -> "activation_record".toJson)
+ val message = Map(
+ "message" -> s"Activation record '${activation.activationId}' for entity
'${activation.name}'".toJson)
+ val annotations = activation.annotations.toJsObject.fields
+ val addFields = transactionType ++ annotations ++ message ++
additionalFields
+ val removeFields = Seq("logs", "annotations")
+ val line = activation.metadata.toExtendedJson(removeFields, addFields)
+
+ ByteString(s"${line.compactPrint}\n")
+ }
+
+ def getLogFile = logFile
+
+ def activationToFile(activation: WhiskActivation,
+ context: UserContext,
+ additionalFields: Map[String, JsValue] = Map.empty) = {
+ val transcribedLogs = transcribeLogs(activation, additionalFields)
+ val transcribedActivation = transcribeActivation(activation,
additionalFields)
+
+ // Write each log line to file and then write the activation metadata
+ Source
+ .fromIterator(() => transcribedLogs.toIterator)
+
.runWith(Flow[ByteString].concat(Source.single(transcribedActivation)).to(writeToFile))
+ }
+}
diff --git
a/common/scala/src/main/scala/org/apache/openwhisk/core/database/ArtifactWithFileStorageActivationStore.scala
b/common/scala/src/main/scala/org/apache/openwhisk/core/database/ArtifactWithFileStorageActivationStore.scala
new file mode 100644
index 0000000..597a1f9
--- /dev/null
+++
b/common/scala/src/main/scala/org/apache/openwhisk/core/database/ArtifactWithFileStorageActivationStore.scala
@@ -0,0 +1,61 @@
+/*
+ * 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 org.apache.openwhisk.core.database
+
+import java.nio.file.Paths
+
+import akka.actor.ActorSystem
+import akka.stream._
+import org.apache.openwhisk.common.{Logging, TransactionId}
+import org.apache.openwhisk.core.ConfigKeys
+import org.apache.openwhisk.core.entity.{DocInfo, _}
+import pureconfig.loadConfigOrThrow
+import spray.json._
+
+import scala.concurrent.Future
+
+case class ArtifactWithFileStorageActivationStoreConfig(logFilePrefix: String,
logPath: String, userIdField: String)
+
+class ArtifactWithFileStorageActivationStore(
+ actorSystem: ActorSystem,
+ actorMaterializer: ActorMaterializer,
+ logging: Logging,
+ config: ArtifactWithFileStorageActivationStoreConfig =
+
loadConfigOrThrow[ArtifactWithFileStorageActivationStoreConfig](ConfigKeys.activationStoreWithFileStorage))
+ extends ArtifactActivationStore(actorSystem, actorMaterializer, logging) {
+
+ private val activationFileStorage =
+ new ActivationFileStorage(config.logFilePrefix, Paths.get(config.logPath),
actorMaterializer, logging)
+
+ def getLogFile = activationFileStorage.getLogFile
+
+ override def store(activation: WhiskActivation, context: UserContext)(
+ implicit transid: TransactionId,
+ notifier: Option[CacheChangeNotification]): Future[DocInfo] = {
+ val additionalFields = Map(config.userIdField ->
context.user.namespace.uuid.toJson)
+
+ activationFileStorage.activationToFile(activation, context,
additionalFields)
+ super.store(activation, context)
+ }
+
+}
+
+object ArtifactWithFileStorageActivationStoreProvider extends
ActivationStoreProvider {
+ override def instance(actorSystem: ActorSystem, actorMaterializer:
ActorMaterializer, logging: Logging) =
+ new ArtifactWithFileStorageActivationStore(actorSystem, actorMaterializer,
logging)
+}
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 59c892a..4222e51 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
@@ -99,9 +99,10 @@ case class WhiskActivation(namespace: EntityPath,
def resultAsJson = response.result.toJson.asJsObject
- def toExtendedJson = {
+ def toExtendedJson(removeFields: Seq[String] = Seq.empty, addFields:
Map[String, JsValue] = Map.empty) = {
val JsObject(baseFields) = WhiskActivation.serdes.write(this).asJsObject
- val newFields = (baseFields - "response") + ("response" ->
response.toExtendedJson)
+
+ val newFields = (baseFields - "response") + ("response" ->
response.toExtendedJson) -- removeFields ++ addFields
if (end != Instant.EPOCH) {
val durationValue = (duration getOrElse (end.toEpochMilli -
start.toEpochMilli)).toJson
JsObject(newFields + ("duration" -> durationValue))
@@ -110,10 +111,17 @@ case class WhiskActivation(namespace: EntityPath,
}
}
+ def metadata = {
+ copy(response = response.withoutResult, annotations = Parameters(), logs =
ActivationLogs())
+ .revision[WhiskActivation](rev)
+ }
+ def withoutResult = {
+ copy(response = response.withoutResult)
+ .revision[WhiskActivation](rev)
+ }
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)
}
diff --git
a/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Actions.scala
b/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Actions.scala
index 3587a7e..d535fd1 100644
---
a/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Actions.scala
+++
b/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Actions.scala
@@ -263,7 +263,7 @@ trait WhiskActionsApi extends WhiskCollectionAPI with
PostActionActivation with
complete(Accepted, activationId.toJsObject)
}
case Success(Right(activation)) =>
- val response = if (result) activation.resultAsJson else
activation.toExtendedJson
+ val response = if (result) activation.resultAsJson else
activation.toExtendedJson()
respondWithActivationIdHeader(activation.activationId) {
if (activation.response.isSuccess) {
diff --git
a/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Activations.scala
b/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Activations.scala
index 5e8f0ab..9100ba4 100644
---
a/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Activations.scala
+++
b/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Activations.scala
@@ -172,7 +172,7 @@ trait WhiskActivationsApi extends Directives with
AuthenticatedRouteProvider wit
case None =>
activationStore.listActivationsInNamespace(namespace, skip.n,
limit.n, docs, since, upto, context)
}
- listEntities(activations map (_.fold((js) => js, (wa) =>
wa.map(_.toExtendedJson))))
+ listEntities(activations map (_.fold((js) => js, (wa) =>
wa.map(_.toExtendedJson()))))
}
}
}
@@ -191,7 +191,7 @@ trait WhiskActivationsApi extends Directives with
AuthenticatedRouteProvider wit
pathEndOrSingleSlash {
getEntity(
activationStore.get(ActivationId(docid.asString), context),
- postProcess = Some((activation: WhiskActivation) =>
complete(activation.toExtendedJson)))
+ postProcess = Some((activation: WhiskActivation) =>
complete(activation.toExtendedJson())))
} ~ (pathPrefix(resultPath) & pathEnd) { fetchResponse(context, docid) } ~
(pathPrefix(logsPath) & pathEnd) { fetchLogs(context, docid) }
}
diff --git
a/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActionsApiTests.scala
b/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActionsApiTests.scala
index b69536b..761a09b 100644
---
a/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActionsApiTests.scala
+++
b/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActionsApiTests.scala
@@ -1386,7 +1386,7 @@ class ActionsApiTests extends ControllerTestCommon with
WhiskActionsApi {
Post(s"$collectionPath/${action.name}?blocking=true") ~>
Route.seal(routes(creds)) ~> check {
status should be(OK)
val response = responseAs[JsObject]
- response should be(activation.withoutLogs.toExtendedJson)
+ response should be(activation.withoutLogs.toExtendedJson())
}
// repeat invoke, get only result back
@@ -1422,7 +1422,7 @@ class ActionsApiTests extends ControllerTestCommon with
WhiskActionsApi {
Post(s"$collectionPath/${action.name}?blocking=true") ~>
Route.seal(routes(creds)) ~> check {
status should be(OK)
val response = responseAs[JsObject]
- response should be(activation.withoutLogs.toExtendedJson)
+ response should be(activation.withoutLogs.toExtendedJson())
}
// repeat invoke, get only result back
@@ -1475,7 +1475,7 @@ class ActionsApiTests extends ControllerTestCommon with
WhiskActionsApi {
Post(s"$collectionPath/${action.name}?blocking=true&timeout=500") ~>
Route.seal(routes(creds)) ~> check {
status shouldBe OK
val response = responseAs[JsObject]
- response shouldBe activation.withoutLogs.toExtendedJson
+ response shouldBe activation.withoutLogs.toExtendedJson()
headers should contain(RawHeader(ActivationIdHeader,
response.fields("activationId").convertTo[String]))
}
} finally {
@@ -1504,7 +1504,7 @@ class ActionsApiTests extends ControllerTestCommon with
WhiskActionsApi {
Post(s"$collectionPath/${action.name}?blocking=true") ~>
Route.seal(routes(creds)) ~> check {
status should be(InternalServerError)
val response = responseAs[JsObject]
- response should be(activation.withoutLogs.toExtendedJson)
+ response should be(activation.withoutLogs.toExtendedJson())
headers should contain(RawHeader(ActivationIdHeader,
response.fields("activationId").convertTo[String]))
}
} finally {
diff --git
a/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActivationsApiTests.scala
b/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActivationsApiTests.scala
index f540c6e..8ad2e4b 100644
---
a/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActivationsApiTests.scala
+++
b/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActivationsApiTests.scala
@@ -188,7 +188,7 @@ class ActivationsApiTests extends ControllerTestCommon with
WhiskActivationsApi
status should be(OK)
val response = responseAs[List[JsObject]]
activations.length should be(response.length)
- response should contain theSameElementsAs
activations.map(_.toExtendedJson)
+ response should contain theSameElementsAs
activations.map(_.toExtendedJson())
}
}
} finally {
@@ -270,7 +270,7 @@ class ActivationsApiTests extends ControllerTestCommon with
WhiskActivationsApi
status should be(OK)
val response = responseAs[List[JsObject]]
expected.length should be(response.length)
- response should contain theSameElementsAs
expected.map(_.toExtendedJson)
+ response should contain theSameElementsAs
expected.map(_.toExtendedJson())
}
}
}
@@ -286,7 +286,7 @@ class ActivationsApiTests extends ControllerTestCommon with
WhiskActivationsApi
status should be(OK)
val response = responseAs[List[JsObject]]
expected.length should be(response.length)
- response should contain theSameElementsAs
expected.map(_.toExtendedJson)
+ response should contain theSameElementsAs
expected.map(_.toExtendedJson())
}
}
}
@@ -302,7 +302,7 @@ class ActivationsApiTests extends ControllerTestCommon with
WhiskActivationsApi
status should be(OK)
val response = responseAs[List[JsObject]]
expected.length should be(response.length)
- response should contain theSameElementsAs
expected.map(_.toExtendedJson)
+ response should contain theSameElementsAs
expected.map(_.toExtendedJson())
}
}
}
@@ -538,14 +538,14 @@ class ActivationsApiTests extends ControllerTestCommon
with WhiskActivationsApi
Get(s"$collectionPath/${activation.activationId.asString}") ~>
Route.seal(routes(creds)) ~> check {
status should be(OK)
val response = responseAs[JsObject]
- response should be(activation.toExtendedJson)
+ response should be(activation.toExtendedJson())
}
// it should "get activation by name in explicit namespace owned by
subject" in
Get(s"/$namespace/${collection.path}/${activation.activationId.asString}") ~>
Route.seal(routes(creds)) ~> check {
status should be(OK)
val response = responseAs[JsObject]
- response should be(activation.toExtendedJson)
+ response should be(activation.toExtendedJson())
}
// it should "reject get activation by name in explicit namespace not
owned by subject" in
diff --git
a/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ControllerTestCommon.scala
b/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ControllerTestCommon.scala
index d623b3d..9e58639 100644
---
a/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ControllerTestCommon.scala
+++
b/tests/src/test/scala/org/apache/openwhisk/core/controller/test/ControllerTestCommon.scala
@@ -125,7 +125,9 @@ protected trait ControllerTestCommon
() => {
val activations: Future[Either[List[JsObject], List[WhiskActivation]]]
=
activationStore.listActivationsInNamespace(namespace, 0, 0, context
= context)
- val listFuture: Future[List[JsObject]] = activations map (_.fold((js)
=> js, (wa) => wa.map(_.toExtendedJson)))
+ val listFuture: Future[List[JsObject]] = activations map (_.fold(
+ (js) => js,
+ (wa) => wa.map(_.toExtendedJson())))
listFuture map { l =>
if (l.length != count) {
@@ -146,7 +148,9 @@ protected trait ControllerTestCommon
() => {
val activations: Future[Either[List[JsObject], List[WhiskActivation]]]
=
activationStore.listActivationsMatchingName(namespace, name, 0, 0,
context = context)
- val listFuture: Future[List[JsObject]] = activations map (_.fold((js)
=> js, (wa) => wa.map(_.toExtendedJson)))
+ val listFuture: Future[List[JsObject]] = activations map (_.fold(
+ (js) => js,
+ (wa) => wa.map(_.toExtendedJson())))
listFuture map { l =>
if (l.length != count) {
diff --git
a/tests/src/test/scala/org/apache/openwhisk/core/database/ArtifactWithFileStorageActivationStoreTests.scala
b/tests/src/test/scala/org/apache/openwhisk/core/database/ArtifactWithFileStorageActivationStoreTests.scala
new file mode 100644
index 0000000..9253844
--- /dev/null
+++
b/tests/src/test/scala/org/apache/openwhisk/core/database/ArtifactWithFileStorageActivationStoreTests.scala
@@ -0,0 +1,157 @@
+/*
+ * 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 org.apache.openwhisk.core.database
+
+import java.io.File
+import java.time.Instant
+
+import akka.actor.ActorSystem
+import akka.http.scaladsl.model.HttpRequest
+import akka.stream.ActorMaterializer
+import akka.testkit.TestKit
+import common.StreamLogging
+import org.junit.runner.RunWith
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.junit.JUnitRunner
+import org.scalatest.{FlatSpecLike, Matchers}
+import spray.json.DefaultJsonProtocol._
+import spray.json._
+import org.apache.openwhisk.common.TransactionId
+import org.apache.openwhisk.core.entity._
+import org.apache.openwhisk.core.entity.size.SizeInt
+import scala.concurrent.{Await, Future}
+import scala.concurrent.duration._
+import scala.io.Source
+
+@RunWith(classOf[JUnitRunner])
+class ArtifactWithFileStorageActivationStoreTests()
+ extends TestKit(ActorSystem("ArtifactWithFileStorageActivationStoreTests"))
+ with FlatSpecLike
+ with Matchers
+ with ScalaFutures
+ with StreamLogging {
+
+ implicit val transid: TransactionId = TransactionId.testing
+ implicit val notifier: Option[CacheChangeNotification] = None
+
+ private val materializer = ActorMaterializer()
+ private val uuid = UUID()
+ private val subject = Subject()
+ private val user =
+ Identity(subject, Namespace(EntityName("testSpace"), uuid),
BasicAuthenticationAuthKey(uuid, Secret()), Set())
+ private val context = UserContext(user, HttpRequest())
+
+ private def await[T](awaitable: Future[T], timeout: FiniteDuration =
10.seconds) = Await.result(awaitable, timeout)
+
+ def responsePermutations = {
+ val message = JsObject("result key" -> JsString("result value"))
+ Seq(
+ ActivationResponse.success(None),
+ ActivationResponse.success(Some(message)),
+ ActivationResponse.applicationError(message),
+ ActivationResponse.whiskError(message))
+ }
+
+ def logPermutations = {
+ Seq(
+ ActivationLogs(),
+ ActivationLogs(Vector("2018-03-05T02:10:38.196689520Z stdout: single log
line")),
+ ActivationLogs(
+ Vector(
+ "2018-03-05T02:10:38.196689522Z stdout: first log line of multiple
lines",
+ "2018-03-05T02:10:38.196754258Z stdout: second log line of multiple
lines")))
+ }
+
+ def expectedFileContent(activation: WhiskActivation) = {
+ val expectedLogs = activation.logs.logs.map { log =>
+ JsObject(
+ "type" -> "user_log".toJson,
+ "message" -> log.toJson,
+ "activationId" -> activation.activationId.toJson,
+ "namespaceId" -> user.namespace.uuid.toJson)
+ }
+ val expectedActivation = JsObject(
+ "type" -> "activation_record".toJson,
+ "duration" -> activation.duration.toJson,
+ "name" -> activation.name.toJson,
+ "subject" -> activation.subject.toJson,
+ "waitTime" -> activation.annotations.get("waitTime").toJson.toJson,
+ "activationId" -> activation.activationId.toJson,
+ "namespaceId" -> user.namespace.uuid.toJson,
+ "publish" -> activation.publish.toJson,
+ "version" -> activation.version.toJson,
+ "response" -> activation.response.withoutResult.toExtendedJson,
+ "end" -> activation.end.toEpochMilli.toJson,
+ "message" -> s"Activation record '${activation.activationId}' for entity
'${activation.name}'".toJson,
+ "kind" -> activation.annotations.get("kind").toJson.toJson,
+ "start" -> activation.start.toEpochMilli.toJson,
+ "limits" -> activation.annotations.get("limits").toJson.toJson,
+ "initTime" -> activation.annotations.get("initTime").toJson,
+ "namespace" -> activation.namespace.toJson)
+
+ expectedLogs ++ Seq(expectedActivation)
+ }
+
+ it should "store activations in artifact store and to file" in {
+ val config = ArtifactWithFileStorageActivationStoreConfig("userlogs",
"logs", "namespaceId")
+ val activationStore = new ArtifactWithFileStorageActivationStore(system,
materializer, logging, config)
+ val logDir = new File(new File(".").getCanonicalPath, config.logPath)
+
+ try {
+ logDir.mkdir
+
+ val activations = responsePermutations.map { response =>
+ logPermutations.map { logs =>
+ val activation = WhiskActivation(
+ namespace = EntityPath(subject.asString),
+ name = EntityName("name"),
+ subject = subject,
+ activationId = ActivationId.generate(),
+ start = Instant.now,
+ end = Instant.now,
+ response = response,
+ logs = logs,
+ duration = Some(101L),
+ annotations = Parameters("kind", "nodejs:6") ++ Parameters(
+ "limits",
+ ActionLimits(TimeLimit(60.second), MemoryLimit(256.MB),
LogLimit(10.MB)).toJson) ++
+ Parameters("waitTime", 16.toJson) ++
+ Parameters("initTime", 44.toJson))
+ val docInfo = await(activationStore.store(activation, context))
+ val fullyQualifiedActivationId = ActivationId(docInfo.id.asString)
+
+ await(activationStore.get(fullyQualifiedActivationId, context))
shouldBe activation
+ await(activationStore.delete(fullyQualifiedActivationId, context))
+ activation
+ }
+ }.flatten
+
+ Source
+ .fromFile(activationStore.getLogFile.toFile.getAbsoluteFile)
+ .getLines
+ .toList
+ .map(_.parseJson)
+ .toJson
+ .convertTo[JsArray] shouldBe
activations.map(expectedFileContent).flatten.toJson.convertTo[JsArray]
+ } finally {
+ activationStore.getLogFile.toFile.getAbsoluteFile.delete
+ logDir.delete
+ }
+ }
+
+}
