Skip to content

improve http graceful termination #4957

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tysonnorris wants to merge 2 commits into
base: master
Choose a base branch
from
Open

improve http graceful termination #4957

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
27b56af
improve http graceful termination
tysonnorris Aug 28, 2020
8143f51
move readyState to class var instead of an object var
tysonnorris Sep 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion common/scala/src/main/resources/application.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,17 @@ kamon {
}
}
}

akka {
# required to allow longer shutdown-termination-limit below
coordinated-shutdown.default-phase-timeout = 61 s
}
whisk {
http {
shutdown-unready-delay = 15 seconds # /ready will return 500 for this amount of time before connection draining starts
Copy link
Copy Markdown
Member

@style95 style95 Nov 14, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems it would wait for 15 seconds, and start the termination phase.
But if there are still client connections connected, are they getting connection closed error?

Copy link
Copy Markdown
Contributor Author

@tysonnorris tysonnorris Nov 19, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This only allows an unready stage, which only affects the added /ready endpoint to return 503 (no other endpoints affected). This does not affect how akka http handles existing client connections during shutdown, so if akka allows gracefully terminating in flight requests, then no connection closed errors, but no new connections allowed. The purpose here is to allow external systems (e.g. kubernetes or reverse proxies) to probe the service and stop routing to this instance for some time before akka shutdown initiates.

shutdown-termination-limit = 61 seconds # hard limit on how long (after unready delay) requests have to complete
}


shared-packages-execute-only = false
metrics {
# Enable/disable Prometheus support. If enabled then metrics would be exposed at `/metrics` endpoint
Expand Down
59 changes: 46 additions & 13 deletions common/scala/src/main/scala/org/apache/openwhisk/http/BasicHttpService.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,8 @@

package org.apache.openwhisk.http

import akka.actor.ActorSystem
import akka.Done
import akka.actor.{ActorSystem, CoordinatedShutdown}
import akka.event.Logging
import akka.http.scaladsl.{Http, HttpConnectionContext}
import akka.http.scaladsl.model.{HttpRequest, _}
Expand All @@ -29,17 +30,23 @@ import kamon.metric.MeasurementUnit
import spray.json._
import org.apache.openwhisk.common.Https.HttpsConfig
import org.apache.openwhisk.common._

import akka.pattern.after
import org.apache.openwhisk.common.TransactionId.systemPrefix
import pureconfig._
import pureconfig.generic.auto._
import scala.collection.immutable.Seq
import scala.concurrent.duration.DurationInt
import scala.concurrent.{Await, Future}
import scala.concurrent.duration.{DurationInt, FiniteDuration}
import scala.concurrent.Future
import scala.util.{Failure, Success}

case class BasicHttpServiceConfig(shutdownUnreadyDelay: FiniteDuration, shutdownTerminationLimit: FiniteDuration)

/**
* This trait extends the Akka Directives and Actor with logging and transaction counting
* facilities common to all OpenWhisk REST services.
*/
trait BasicHttpService extends Directives {

implicit val logging: Logging
val OW_EXTRA_LOGGING_HEADER = "X-OW-EXTRA-LOGGING"

/**
Expand Down Expand Up @@ -156,24 +163,50 @@ trait BasicHttpService extends Directives {
}

object BasicHttpService {
implicit val tid = TransactionId(systemPrefix + "http_service")
//start with ready true
protected[http] var ready = true
Copy link
Copy Markdown
Member

@rabbah rabbah Aug 29, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this used for anything?

Copy link
Copy Markdown
Contributor Author

@tysonnorris tysonnorris Aug 31, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The ready var? It is used to alter the response to /ready endpoint defined in BasicRasService.
I'll try to refactor this so that we're not modifying the object var from different places.


/**
* Starts an HTTP(S) route handler on given port and registers a shutdown hook.
*/
def startHttpService(route: Route, port: Int, config: Option[HttpsConfig] = None, interface: String = "0.0.0.0")(
def startHttpService(route: Route, port: Int, httpsConfig: Option[HttpsConfig] = None, interface: String = "0.0.0.0")(
implicit actorSystem: ActorSystem,
materializer: ActorMaterializer): Unit = {
val connectionContext = config.map(Https.connectionContext(_)).getOrElse(HttpConnectionContext)
materializer: ActorMaterializer,
logging: Logging): Unit = {
val connectionContext = httpsConfig.map(Https.connectionContext(_)).getOrElse(HttpConnectionContext)
val httpBinding = Http().bindAndHandle(route, interface, port, connectionContext = connectionContext)
logging.info(this, "starting http service...")
addShutdownHook(httpBinding)
}

def addShutdownHook(binding: Future[Http.ServerBinding])(implicit actorSystem: ActorSystem,
materializer: ActorMaterializer): Unit = {
def addShutdownHook(binding: Future[Http.ServerBinding],
httpServiceConfig: BasicHttpServiceConfig =
loadConfigOrThrow[BasicHttpServiceConfig]("whisk.http"))(implicit actorSystem: ActorSystem,
materializer: ActorMaterializer,
logging: Logging): Unit = {
implicit val executionContext = actorSystem.dispatcher
sys.addShutdownHook {
Await.result(binding.map(_.unbind()), 30.seconds)
Await.result(actorSystem.whenTerminated, 30.seconds)

CoordinatedShutdown(actorSystem).addTask(CoordinatedShutdown.PhaseBeforeServiceUnbind, "http_unready") { () =>
logging.info(this, "shutdown unready...")
//return 503 status at /ready endpoint for some time before actual termination begins
ready = false
after(httpServiceConfig.shutdownUnreadyDelay, actorSystem.scheduler) {
logging.info(this, "shutdown unready complete...")
Future.successful(Done)
}
}
CoordinatedShutdown(actorSystem).addTask(CoordinatedShutdown.PhaseServiceUnbind, "http_termination") { () =>
logging.info(this, "shutdown terminating...")
binding
.flatMap(_.terminate(hardDeadline = httpServiceConfig.shutdownTerminationLimit))
.andThen {
case Success(_) => logging.info(this, "shutdown termination complete...")
case Failure(t) => logging.info(this, s"shutdown termination failed... ${t}")
}
.map { _ =>
Done
}
}
}

Expand Down
17 changes: 14 additions & 3 deletions common/scala/src/main/scala/org/apache/openwhisk/http/BasicRasService.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,16 @@
package org.apache.openwhisk.http

import akka.event.Logging
import org.apache.openwhisk.common.{MetricsRoute, TransactionId}
import akka.http.scaladsl.model.StatusCodes
import org.apache.openwhisk.common.{Logging, MetricsRoute, TransactionId}

/**
* This trait extends the BasicHttpService with a standard "ping" endpoint which
* responds to health queries, intended for monitoring.
*/
trait BasicRasService extends BasicHttpService {
class BasicRasService(implicit val logging: Logging) extends BasicHttpService {

override def routes(implicit transid: TransactionId) = ping ~ MetricsRoute()
override def routes(implicit transid: TransactionId) = ping ~ ready ~ MetricsRoute()

override def loglevelForRoute(route: String): Logging.LogLevel = {
if (route == "/ping" || route == "/metrics") {
Expand All @@ -39,4 +40,14 @@ trait BasicRasService extends BasicHttpService {
val ping = path("ping") {
get { complete("pong") }
}
val ready = path("ready") {
get {
if (BasicHttpService.ready) {
complete("ok")
} else {
logging.warn(this, "not ready...")
Copy link
Copy Markdown
Contributor

@bdoyle0182 bdoyle0182 Feb 16, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can this just be a debug? I think ready route returning 503 should be self explanatory

complete(StatusCodes.ServiceUnavailable, "notready")
}
}
}
}
5 changes: 3 additions & 2 deletions core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ class Controller(val instance: ControllerInstanceId,
implicit val whiskConfig: WhiskConfig,
implicit val actorSystem: ActorSystem,
implicit val materializer: ActorMaterializer,
implicit val logging: Logging)
override implicit val logging: Logging)
extends BasicRasService {

TransactionId.controller.mark(
Expand Down Expand Up @@ -288,7 +288,8 @@ object Controller {

BasicHttpService.startHttpService(controller.route, port, httpsConfig, interface)(
actorSystem,
controller.materializer)
controller.materializer,
logger)

case Failure(t) =>
abort(s"Invalid runtimes manifest: $t")
Expand Down
3 changes: 2 additions & 1 deletion core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,8 @@ object Invoker {
val invokerServer = SpiLoader.get[InvokerServerProvider].instance(invoker)
BasicHttpService.startHttpService(invokerServer.route, port, httpsConfig)(
actorSystem,
ActorMaterializer.create(actorSystem))
ActorMaterializer.create(actorSystem),
logger)
}
}

Expand Down
8 changes: 6 additions & 2 deletions core/standalone/src/main/scala/org/apache/openwhisk/standalone/PlaygroundLauncher.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import akka.stream.scaladsl.{Sink, Source}
import ch.megard.akka.http.cors.scaladsl.CorsDirectives._
import org.apache.commons.io.IOUtils
import org.apache.commons.lang3.SystemUtils
import org.apache.openwhisk.common.{Logging, TransactionId}
import org.apache.openwhisk.common.{AkkaLogging, Logging, TransactionId}
import org.apache.openwhisk.core.ExecManifestSupport
import org.apache.openwhisk.http.BasicHttpService
import pureconfig._
Expand Down Expand Up @@ -75,7 +75,10 @@ class PlaygroundLauncher(host: String,
private val wsk = new Wsk(host, controllerPort, authKey)

def run(): ServiceContainer = {
BasicHttpService.startHttpService(PlaygroundService.route, pgPort, None, interface)(actorSystem, materializer)
BasicHttpService.startHttpService(PlaygroundService.route, pgPort, None, interface)(
actorSystem,
materializer,
logging)
ServiceContainer(pgPort, pgUrl, "Playground")
}

Expand Down Expand Up @@ -119,6 +122,7 @@ class PlaygroundLauncher(host: String,
}

object PlaygroundService extends BasicHttpService {
override implicit val logging = new AkkaLogging(actorSystem.log)
override def routes(implicit transid: TransactionId): Route =
path(PathEnd | Slash | pg) { redirect(s"/$pg/ui/index.html", StatusCodes.Found) } ~
cors() {
Expand Down