Limit response size.
Move HTTP util to container package where it is actually used (it is no longer used elsewhere).
Remove a lot of now dead code from HTTP util.
Remove connection pool since there is at most one connection to a container at any given time.
Changed container response from Option(int, array of bytes) to Either(container error, container result)
with more useful typing.
Detect larger than allowed response from container and truncate it properly and wrap with container error.
Some logging tweaks.
diff --git a/tests/build.gradle b/tests/build.gradle
index 5378a6a..dd2332e 100644
--- a/tests/build.gradle
+++ b/tests/build.gradle
@@ -52,6 +52,7 @@
testCompile "org.scala-lang:scala-library:${gradle.scala.version}"
testCompile 'org.apache.commons:commons-exec:1.1'
testCompile 'org.apache.commons:commons-lang3:3.3.2'
+ testCompile 'org.apache.httpcomponents:httpclient:4.5.2:tests'
testCompile 'commons-logging:commons-logging:1.1.3'
testCompile 'org.codehaus.groovy:groovy:2.4.3'
testCompile 'org.codehaus.groovy:groovy-json:2.4.3'
diff --git a/tests/src/actionContainers/ActionContainer.scala b/tests/src/actionContainers/ActionContainer.scala
index 1937b3f..c591b52 100644
--- a/tests/src/actionContainers/ActionContainer.scala
+++ b/tests/src/actionContainers/ActionContainer.scala
@@ -155,7 +155,7 @@
private def syncPost(host: String, port: Int, endPoint: String, content: JsValue)(
implicit actorSystem: ActorSystem): (Int, Option[JsObject]) = {
- import whisk.common.NewHttpUtils
+ import whisk.core.container.AkkaHttpUtils
import akka.http.scaladsl.model._
import akka.http.scaladsl.marshalling._
@@ -173,7 +173,7 @@
val f = for (
entity <- Marshal(content).to[MessageEntity];
request = HttpRequest(method = HttpMethods.POST, uri = uri, entity = entity);
- response <- NewHttpUtils.singleRequest(request, 30.seconds, retryOnTCPErrors = true);
+ response <- AkkaHttpUtils.singleRequest(request, 30.seconds, retryOnTCPErrors = true);
responseBody <- Unmarshal(response.entity).to[String]
) yield (response.status.intValue, Try(responseBody.parseJson.asJsObject).toOption)
diff --git a/tests/src/whisk/core/container/test/ContainerConnectionTests.scala b/tests/src/whisk/core/container/test/ContainerConnectionTests.scala
new file mode 100644
index 0000000..13befcf
--- /dev/null
+++ b/tests/src/whisk/core/container/test/ContainerConnectionTests.scala
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2015-2016 IBM Corporation
+ *
+ * Licensed 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.container.test
+
+import java.time.Instant
+import java.nio.charset.StandardCharsets
+
+import scala.concurrent.duration._
+
+import org.junit.runner.RunWith
+import org.scalatest.BeforeAndAfter
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.FlatSpec
+import org.scalatest.Matchers
+import org.scalatest.junit.JUnitRunner
+
+import org.apache.http.localserver.LocalServerTestBase
+import org.apache.http.protocol.HttpRequestHandler
+import org.apache.http.HttpResponse
+import org.apache.http.HttpRequest
+import org.apache.http.protocol.HttpContext
+import org.apache.http.entity.StringEntity
+
+import spray.json.JsObject
+
+import whisk.core.container.HttpUtils
+import whisk.core.entity.size._
+import whisk.core.entity.ActivationResponse._
+
+/**
+ * Unit tests for HttpUtils which communicate with containers.
+ */
+@RunWith(classOf[JUnitRunner])
+class ContainerConnectionTests
+ extends FlatSpec
+ with Matchers
+ with BeforeAndAfter
+ with BeforeAndAfterAll {
+
+ var testHang: FiniteDuration = 0.second
+ var testStatusOK: Boolean = true
+ var testResponse: String = null
+
+ val mockServer = new LocalServerTestBase {
+ override def setUp() = {
+ super.setUp()
+ this.serverBootstrap.registerHandler("/init", new HttpRequestHandler() {
+ override def handle(request: HttpRequest, response: HttpResponse, context: HttpContext) = {
+ if (testHang.length > 0) {
+ Thread.sleep(testHang.toMillis)
+ }
+ response.setStatusCode(if (testStatusOK) 200 else 500);
+ if (testResponse != null) {
+ response.setEntity(new StringEntity(testResponse, StandardCharsets.UTF_8))
+ }
+ }
+ })
+ }
+ }
+
+ mockServer.setUp()
+ val httpHost = mockServer.start()
+ val hostWithPort = s"${httpHost.getHostName}:${httpHost.getPort}"
+
+ before {
+ testHang = 0.second
+ testStatusOK = true
+ testResponse = null
+ }
+
+ override def afterAll = {
+ mockServer.shutDown()
+ }
+
+ behavior of "Container HTTP Utils"
+
+ it should "not wait longer than set timeout" in {
+ val timeout = 5.seconds
+ val connection = new HttpUtils(hostWithPort, timeout, 1.B)
+ testHang = timeout * 2
+ val start = Instant.now()
+ val result = connection.post("/init", JsObject())
+ val end = Instant.now()
+ val waited = end.toEpochMilli - start.toEpochMilli
+ result.isLeft shouldBe true
+ waited should be > timeout.toMillis
+ waited should be < (timeout * 2).toMillis
+ }
+
+ it should "not truncate responses within limit" in {
+ val timeout = 1.minute.toMillis
+ val connection = new HttpUtils(hostWithPort, timeout.millis, 50.B)
+ Seq(true, false).foreach { code =>
+ Seq(null, "", "abc", """{"a":"B"}""", """["a", "b"]""").foreach { r =>
+ testStatusOK = code
+ testResponse = r
+ val result = connection.post("/init", JsObject())
+ result shouldBe Right {
+ ContainerResponse(okStatus = testStatusOK, if (r != null) r else "", None)
+ }
+ }
+ }
+ }
+
+ it should "truncate responses that exceed limit" in {
+ val timeout = 1.minute.toMillis
+ val limit = 1.B
+ val excess = limit + 1.B
+ val connection = new HttpUtils(hostWithPort, timeout.millis, limit)
+ Seq(true, false).foreach { code =>
+ Seq("abc", """{"a":"B"}""", """["a", "b"]""").foreach { r =>
+ testStatusOK = code
+ testResponse = r
+ val result = connection.post("/init", JsObject())
+ result shouldBe Right {
+ ContainerResponse(okStatus = testStatusOK, r.take(limit.toBytes.toInt), Some((r.length.B, limit)))
+ }
+ }
+ }
+ }
+}
diff --git a/tests/src/whisk/core/container/test/ContainerPoolTests.scala b/tests/src/whisk/core/container/test/ContainerPoolTests.scala
index afb5474..f924db3 100644
--- a/tests/src/whisk/core/container/test/ContainerPoolTests.scala
+++ b/tests/src/whisk/core/container/test/ContainerPoolTests.scala
@@ -48,7 +48,6 @@
/**
* Unit tests for ContainerPool and, by association, Container and WhiskContainer.
- *
*/
@RunWith(classOf[JUnitRunner])
class ContainerPoolTests extends FlatSpec
diff --git a/tests/src/whisk/core/dispatcher/test/ActivationResponseTests.scala b/tests/src/whisk/core/dispatcher/test/ActivationResponseTests.scala
index 66777b2..ccf7673 100644
--- a/tests/src/whisk/core/dispatcher/test/ActivationResponseTests.scala
+++ b/tests/src/whisk/core/dispatcher/test/ActivationResponseTests.scala
@@ -28,6 +28,7 @@
import whisk.common.Logging
import whisk.core.entity.ActivationResponse._
import whisk.http.Messages._
+import whisk.core.entity.size.SizeInt
@RunWith(classOf[JUnitRunner])
class ActivationResponseTests extends FlatSpec with Matchers {
@@ -36,131 +37,155 @@
val logger = new Logging {}
+ it should "interpret truncated response" in {
+ val max = 5.B
+ Seq("abcdef", """{"msg":"abcedf"}""", """["a","b","c","d","e"]""").foreach { m =>
+ {
+ val response = ContainerResponse(okStatus = false, m.take(max.toBytes.toInt - 1), Some(m.length.B, max))
+ val init = processInitResponseContent(Right(response), logger)
+ init.statusCode shouldBe ContainerError
+ init.result.get.asJsObject.fields(ERROR_FIELD) shouldBe truncatedResponse(response.entity, m.length.B, max).toJson
+ }
+ {
+ val response = ContainerResponse(okStatus = true, m.take(max.toBytes.toInt - 1), Some(m.length.B, max))
+ val run = processRunResponseContent(Right(response), logger)
+ run.statusCode shouldBe ContainerError
+ run.result.get.asJsObject.fields(ERROR_FIELD) shouldBe truncatedResponse(response.entity, m.length.B, max).toJson
+ }
+ }
+ }
+
it should "interpret failed init that does not response" in {
- val ar = processInitResponseContent(None, logger)
- ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe abnormalInitialization.toJson
+ Seq(NoHost(), ConnectionError(new Throwable()), NoResponseReceived(), Timeout())
+ .map(Left(_)).foreach { e =>
+ val ar = processInitResponseContent(e, logger)
+ ar.statusCode shouldBe ContainerError
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe abnormalInitialization.toJson
+ }
}
it should "interpret failed init that responds with null string" in {
- val response = Some(500, null)
- val ar = processInitResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, null)
+ val ar = processInitResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.get._2).toJson
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.entity).toJson
ar.result.get.toString should not include regex("null")
}
it should "interpret failed init that responds with empty string" in {
- val response = Some(500, "")
- val ar = processInitResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, "")
+ val ar = processInitResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.get._2).toJson
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.entity).toJson
ar.result.get.asJsObject.fields(ERROR_FIELD).toString.endsWith(".\"") shouldBe true
}
it should "interpret failed init that responds with non-empty string" in {
- val response = Some(500, "true")
- val ar = processInitResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, "string")
+ val ar = processInitResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.get._2).toJson
- ar.result.get.toString should include(response.get._2)
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.entity).toJson
+ ar.result.get.toString should include(response.entity)
}
it should "interpret failed init that responds with JSON string not object" in {
- val response = Some(500, Vector(1).toJson.compactPrint)
- val ar = processInitResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, Vector(1).toJson.compactPrint)
+ val ar = processInitResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.get._2).toJson
- ar.result.get.toString should include(response.get._2)
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.entity).toJson
+ ar.result.get.toString should include(response.entity)
}
it should "interpret failed init that responds with JSON object containing error" in {
- val response = Some(500, Map(ERROR_FIELD -> "foobar").toJson.compactPrint)
- val ar = processInitResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, Map(ERROR_FIELD -> "foobar").toJson.compactPrint)
+ val ar = processInitResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get shouldBe response.get._2.parseJson
+ ar.result.get shouldBe response.entity.parseJson
}
it should "interpret failed init that responds with JSON object" in {
- val response = Some(500, Map("foobar" -> "baz").toJson.compactPrint)
- val ar = processInitResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, Map("foobar" -> "baz").toJson.compactPrint)
+ val ar = processInitResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.get._2).toJson
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidInitResponse(response.entity).toJson
ar.result.get.toString should include("baz")
}
it should "not interpret successful init" in {
- val response = Some(200, "")
+ val response = ContainerResponse(okStatus = true, "")
an[IllegalArgumentException] should be thrownBy {
- processInitResponseContent(response, logger)
+ processInitResponseContent(Right(response), logger)
}
}
it should "interpret failed run that does not response" in {
- val ar = processRunResponseContent(None, logger)
- ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe abnormalRun.toJson
+ Seq(NoHost(), ConnectionError(new Throwable()), NoResponseReceived(), Timeout())
+ .map(Left(_)).foreach { e =>
+ val ar = processRunResponseContent(e, logger)
+ ar.statusCode shouldBe ContainerError
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe abnormalRun.toJson
+ }
}
it should "interpret failed run that responds with null string" in {
- val response = Some(500, null)
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, null)
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.get._2).toJson
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.entity).toJson
ar.result.get.toString should not include regex("null")
}
it should "interpret failed run that responds with empty string" in {
- val response = Some(500, "")
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, "")
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.get._2).toJson
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.entity).toJson
ar.result.get.asJsObject.fields(ERROR_FIELD).toString.endsWith(".\"") shouldBe true
}
it should "interpret failed run that responds with non-empty string" in {
- val response = Some(500, "true")
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, "string")
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.get._2).toJson
- ar.result.get.toString should include(response.get._2)
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.entity).toJson
+ ar.result.get.toString should include(response.entity)
}
it should "interpret failed run that responds with JSON string not object" in {
- val response = Some(500, Vector(1).toJson.compactPrint)
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, Vector(1).toJson.compactPrint)
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.get._2).toJson
- ar.result.get.toString should include(response.get._2)
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.entity).toJson
+ ar.result.get.toString should include(response.entity)
}
it should "interpret failed run that responds with JSON object containing error" in {
- val response = Some(500, Map(ERROR_FIELD -> "foobar").toJson.compactPrint)
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, Map(ERROR_FIELD -> "foobar").toJson.compactPrint)
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get shouldBe response.get._2.parseJson
+ ar.result.get shouldBe response.entity.parseJson
}
it should "interpret failed run that responds with JSON object" in {
- val response = Some(500, Map("foobar" -> "baz").toJson.compactPrint)
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = false, Map("foobar" -> "baz").toJson.compactPrint)
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ContainerError
- ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.get._2).toJson
+ ar.result.get.asJsObject.fields(ERROR_FIELD) shouldBe invalidRunResponse(response.entity).toJson
ar.result.get.toString should include("baz")
}
it should "interpret successful run that responds with JSON object containing error" in {
- val response = Some(200, Map(ERROR_FIELD -> "foobar").toJson.compactPrint)
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = true, Map(ERROR_FIELD -> "foobar").toJson.compactPrint)
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe ApplicationError
- ar.result.get shouldBe response.get._2.parseJson
+ ar.result.get shouldBe response.entity.parseJson
}
it should "interpret successful run that responds with JSON object" in {
- val response = Some(200, Map("foobar" -> "baz").toJson.compactPrint)
- val ar = processRunResponseContent(response, logger)
+ val response = ContainerResponse(okStatus = true, Map("foobar" -> "baz").toJson.compactPrint)
+ val ar = processRunResponseContent(Right(response), logger)
ar.statusCode shouldBe Success
- ar.result.get shouldBe response.get._2.parseJson
+ ar.result.get shouldBe response.entity.parseJson
}
}
diff --git a/tests/src/whisk/core/limits/ActionLimitsTests.scala b/tests/src/whisk/core/limits/ActionLimitsTests.scala
index fc37f43..3e88e52 100644
--- a/tests/src/whisk/core/limits/ActionLimitsTests.scala
+++ b/tests/src/whisk/core/limits/ActionLimitsTests.scala
@@ -34,10 +34,11 @@
import common.WskTestHelpers
import spray.json._
import spray.json.DefaultJsonProtocol._
-import spray.json.pimpAny
+import whisk.core.entity.ActivationEntityLimit
+import whisk.core.entity.ActivationResponse
import whisk.core.entity.Exec
import whisk.core.entity.LogLimit
-import whisk.core.entity.size.SizeInt
+import whisk.core.entity.size._
import whisk.core.entity.size.SizeString
import whisk.http.Messages
@@ -113,6 +114,29 @@
}
}
+ it should "succeed but truncate result, if result exceeds its limit" in withAssetCleaner(wskprops) {
+ (wp, assetHelper) =>
+ val name = "TestActionCausingExcessiveResult"
+ assetHelper.withCleaner(wsk.action, name) {
+ val actionName = TestUtils.getTestActionFilename("sizedResult.js")
+ (action, _) => action.create(name, Some(actionName))
+ }
+
+ val allowedSize = ActivationEntityLimit.MAX_ACTIVATION_ENTITY_LIMIT.toBytes
+ val run = wsk.action.invoke(name, Map("size" -> (allowedSize + 1).toJson, "char" -> "a".toJson))
+ withActivation(wsk.activation, run) { activation =>
+ val response = activation.response
+ response.success shouldBe false
+ response.status shouldBe ActivationResponse.messageForCode(ActivationResponse.ContainerError)
+ val msg = response.result.get.fields(ActivationResponse.ERROR_FIELD).convertTo[String]
+ val expected = Messages.truncatedResponse((allowedSize + 10).B, allowedSize.B)
+ withClue(s"is: ${msg.take(expected.length)}\nexpected: $expected") {
+ msg.startsWith(expected) shouldBe true
+ }
+ msg.endsWith("a") shouldBe true
+ }
+ }
+
it should "succeed with one log line" in withAssetCleaner(wskprops) {
(wp, assetHelper) =>
val name = "TestActionCausingExceededLogs"
