Controller endpoint for checking the controller readiness (#4858)

* Provide readiness endpoint for controller

* Fix test fmt

* Controller route tests

* Separate unit and integration tests

* Scala fmt

* Adding PR suggested changes

Author: selfxp

core/controller/src/main/resources/reference.conf

@@ -33,5 +33,6 @@ whisk {
   controller {
     protocol: http
     interface: "0.0.0.0"
+    readiness-fraction: 100%
   }
 }

core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala

@@ -21,6 +21,7 @@ import akka.Done
 import akka.actor.{ActorSystem, CoordinatedShutdown}
 import akka.event.Logging.InfoLevel
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
+import akka.http.scaladsl.model.StatusCodes._
 import akka.http.scaladsl.model.Uri
 import akka.http.scaladsl.server.Route
 import akka.stream.ActorMaterializer
@@ -131,12 +132,14 @@ class Controller(val instance: ControllerInstanceId,
   private val swagger = new SwaggerDocs(Uri.Path.Empty, "infoswagger.json")
 
   /**
-   * Handles GET /invokers
-   *             /invokers/healthy/count
+   * Handles GET /invokers - list of invokers
+   *             /invokers/healthy/count - nr of healthy invokers
+   *             /invokers/ready - 200 in case # of healthy invokers are above the expected value
+   *                             - 500 in case # of healthy invokers are bellow the expected value
    *
    * @return JSON with details of invoker health or count of healthy invokers respectively.
    */
-  private val internalInvokerHealth = {
+  protected[controller] val internalInvokerHealth = {
     implicit val executionContext = actorSystem.dispatcher
     (pathPrefix("invokers") & get) {
       pathEndOrSingleSlash {
@@ -151,6 +154,16 @@ class Controller(val instance: ControllerInstanceId,
             .invokerHealth()
             .map(_.count(_.status == InvokerState.Healthy).toJson)
         }
+      } ~ path("ready") {
+        onSuccess(loadBalancer.invokerHealth()) { invokersHealth =>
+          val all = invokersHealth.size
+          val healthy = invokersHealth.count(_.status == InvokerState.Healthy)
+          val ready = Controller.readyState(all, healthy, Controller.readinessThreshold.getOrElse(1))
+          if (ready)
+            complete(JsObject("healthy" -> s"$healthy/$all".toJson))
+          else
+            complete(InternalServerError -> JsObject("unhealthy" -> s"${all - healthy}/$all".toJson))
+        }
       }
     }
   }
@@ -172,6 +185,7 @@ object Controller {
 
   protected val protocol = loadConfigOrThrow[String]("whisk.controller.protocol")
   protected val interface = loadConfigOrThrow[String]("whisk.controller.interface")
+  protected val readinessThreshold = loadConfig[Double]("whisk.controller.readiness-fraction")
 
   // requiredProperties is a Map whose keys define properties that must be bound to
   // a value, and whose values are default values.   A null value in the Map means there is
@@ -207,6 +221,10 @@ object Controller {
         "max_action_logs" -> logLimit.max.toBytes.toJson),
       "runtimes" -> runtimes.toJson)
 
+  def readyState(allInvokers: Int, healthyInvokers: Int, readinessThreshold: Double): Boolean = {
+    if (allInvokers > 0) (healthyInvokers / allInvokers) >= readinessThreshold else false
+  }
+
   def main(args: Array[String]): Unit = {
     implicit val actorSystem = ActorSystem("controller-actor-system")
     implicit val logger = new AkkaLogging(akka.event.Logging.getLogger(actorSystem, this))

tests/src/test/scala/org/apache/openwhisk/core/controller/test/ControllerApiTests.scala

@@ -17,16 +17,15 @@
 
 package org.apache.openwhisk.core.controller.test
 
-import org.junit.runner.RunWith
-import org.scalatest.FlatSpec
-import org.scalatest.Matchers
-import org.scalatest.junit.JUnitRunner
-import io.restassured.RestAssured
 import common.StreamLogging
-import spray.json._
-import spray.json.DefaultJsonProtocol._
+import io.restassured.RestAssured
 import org.apache.openwhisk.core.WhiskConfig
 import org.apache.openwhisk.core.entity.{ExecManifest, LogLimit, MemoryLimit, TimeLimit}
+import org.junit.runner.RunWith
+import org.scalatest.{FlatSpec, Matchers}
+import org.scalatest.junit.JUnitRunner
+import spray.json.DefaultJsonProtocol._
+import spray.json._
 import system.rest.RestUtil
 
 /**
@@ -68,4 +67,27 @@ class ControllerApiTests extends FlatSpec with RestUtil with Matchers with Strea
     response.body.asString.parseJson shouldBe (expectedJson)
   }
 
+  behavior of "Controller"
+
+  it should "return list of invokers" in {
+    val response = RestAssured.given.config(sslconfig).get(s"$getServiceURL/invokers")
+
+    response.statusCode shouldBe 200
+    response.body.asString shouldBe "{\"invoker0/0\":\"up\",\"invoker1/1\":\"up\"}"
+  }
+
+  it should "return healthy invokers status" in {
+    val response = RestAssured.given.config(sslconfig).get(s"$getServiceURL/invokers/healthy/count")
+
+    response.statusCode shouldBe 200
+    response.body.asString shouldBe "2"
+  }
+
+  it should "return healthy invokers" in {
+    val response = RestAssured.given.config(sslconfig).get(s"$getServiceURL/invokers/ready")
+
+    response.statusCode shouldBe 200
+    response.body.asString shouldBe "{\"healthy\":\"2/2\"}"
+  }
+
 }

tests/src/test/scala/org/apache/openwhisk/core/controller/test/ControllerRoutesTests.scala

@@ -0,0 +1,81 @@
+/*
+ * 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.controller.test
+
+import akka.http.scaladsl.model.StatusCodes._
+import akka.http.scaladsl.server.Route
+import org.apache.openwhisk.common.AkkaLogging
+import org.apache.openwhisk.core.controller.Controller
+import org.apache.openwhisk.core.entity.ExecManifest.Runtimes
+import org.junit.runner.RunWith
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.junit.JUnitRunner
+import system.rest.RestUtil
+import spray.json._
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
+import spray.json.DefaultJsonProtocol._
+
+/**
+ * Tests controller readiness.
+ *
+ * These tests will validate the endpoints that provide information on how many healthy invokers are available
+ */
+
+@RunWith(classOf[JUnitRunner])
+class ControllerRoutesTests extends ControllerTestCommon with BeforeAndAfterEach with RestUtil {
+
+  implicit val logger = new AkkaLogging(akka.event.Logging.getLogger(actorSystem, this))
+
+  behavior of "Controller"
+
+  it should "return unhealthy invokers status" in {
+
+    configureBuildInfo()
+
+    val controller =
+      new Controller(instance, Runtimes(Set.empty, Set.empty, None), whiskConfig, system, materializer, logger)
+    Get("/invokers/ready") ~> Route.seal(controller.internalInvokerHealth) ~> check {
+      status shouldBe InternalServerError
+      responseAs[JsObject].fields("unhealthy") shouldBe JsString("0/0")
+    }
+  }
+
+  it should "return ready state true when healthy == total invokers" in {
+
+    val res = Controller.readyState(5, 5, 1.0)
+    res shouldBe true
+  }
+
+  it should "return ready state false when 0 invokers" in {
+
+    val res = Controller.readyState(0, 0, 0.5)
+    res shouldBe false
+  }
+
+  it should "return ready state false when threshold < (healthy / total)" in {
+
+    val res = Controller.readyState(7, 3, 0.5)
+    res shouldBe false
+  }
+
+  private def configureBuildInfo(): Unit = {
+    System.setProperty("whisk.info.build-no", "")
+    System.setProperty("whisk.info.date", "")
+  }
+
+}