Support graceful shutdown. (#5283)

Author: style95

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/Container.scala

@@ -135,6 +135,13 @@ trait Container {
           transid.failed(this, start, s"initializiation failed with $t")
       }
       .flatMap { result =>
+        // if runtime container is shutting down, reschedule the activation message
+        result.response.right.map { res =>
+          if (res.shuttingDown) {
+            throw ContainerHealthError(transid, containerId.asString)
+          }
+        }
+
         if (result.ok) {
           Future.successful(result.interval)
         } else if (result.interval.duration >= timeout) {
@@ -180,6 +187,13 @@ trait Container {
           transid.failed(this, start, s"run failed with $t")
       }
       .map { result =>
+        // if runtime container is shutting down, reschedule the activation message
+        result.response.right.map { res =>
+          if (res.shuttingDown) {
+            throw ContainerHealthError(transid, containerId.asString)
+          }
+        }
+
         val response = if (result.interval.duration >= timeout) {
           ActivationResponse.developerError(Messages.timedoutActivation(timeout, false))
         } else {

common/scala/src/main/scala/org/apache/openwhisk/core/entity/ActivationResult.scala

@@ -18,12 +18,9 @@
 package org.apache.openwhisk.core.entity
 
 import scala.util.Try
-
-import akka.http.scaladsl.model.StatusCodes.OK
-
+import akka.http.scaladsl.model.StatusCodes.{OK, ServiceUnavailable}
 import spray.json._
 import spray.json.DefaultJsonProtocol
-
 import org.apache.openwhisk.common.Logging
 import org.apache.openwhisk.http.Messages._
 
@@ -139,6 +136,10 @@ protected[core] object ActivationResponse extends DefaultJsonProtocol {
     /** true iff status code is OK (HTTP 200 status code), anything else is considered an error. **/
     val okStatus = statusCode == OK.intValue
     val ok = okStatus && truncated.isEmpty
+
+    /** true iff status code is ServiceUnavailable (HTTP 503 status code) */
+    val shuttingDown = statusCode == ServiceUnavailable.intValue
+
     override def toString = {
       val base = if (okStatus) "ok" else "not ok"
       val rest = truncated.map(e => s", truncated ${e.toString}").getOrElse("")

tests/src/test/scala/org/apache/openwhisk/core/containerpool/docker/test/DockerContainerTests.scala

@@ -474,6 +474,23 @@ class DockerContainerTests
     end.deltaToMarkerStart shouldBe Some(interval.duration.toMillis)
   }
 
+  it should "throw ContainerHealthError if runtime container returns 503 response" in {
+    implicit val docker = stub[DockerApiWithFileAccess]
+    implicit val runc = stub[RuncApi]
+
+    val interval = intervalOf(1.millisecond)
+    val result = JsObject.empty
+    val container = dockerContainer() {
+      Future.successful(RunResult(interval, Right(ContainerResponse(503, result.compactPrint, None))))
+    }
+
+    val initResult = container.initialize(JsObject.empty, 1.second, 1)
+    an[ContainerHealthError] should be thrownBy await(initResult)
+
+    val runResult = container.run(JsObject.empty, JsObject.empty, 1.second, 1)
+    an[ContainerHealthError] should be thrownBy await(runResult)
+  }
+
   it should "properly deal with a timeout during run" in {
     implicit val docker = stub[DockerApiWithFileAccess]
     implicit val runc = stub[RuncApi]

tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/KubernetesContainerTests.scala

@@ -277,6 +277,22 @@ class KubernetesContainerTests
     end.deltaToMarkerStart shouldBe Some(interval.duration.toMillis)
   }
 
+  it should "throw ContainerHealthError if runtime container returns 503 response" in {
+    implicit val kubernetes = stub[KubernetesApi]
+
+    val interval = intervalOf(1.millisecond)
+    val result = JsObject.empty
+    val container = kubernetesContainer() {
+      Future.successful(RunResult(interval, Right(ContainerResponse(503, result.compactPrint, None))))
+    }
+
+    val initResult = container.initialize(JsObject.empty, 1.second, 1)
+    an[ContainerHealthError] should be thrownBy await(initResult)
+
+    val runResult = container.run(JsObject.empty, JsObject.empty, 1.second, 1)
+    an[ContainerHealthError] should be thrownBy await(runResult)
+  }
+
   it should "properly deal with a timeout during run" in {
     implicit val kubernetes = stub[KubernetesApi]