Prepare to integrate new scheduler into apache/openwhisk-deploy-kube (#5278)

* Kubernetes Akka bootstrap for controller

* Update cluster management for the scheduler to help with k8s deployment

* Made changes to tools to try to integrate scheduler into travis/CI build
process

* Added scheduler Dockerfile.cov

* Use consistent ordering of components

* remove canonical.port setting in scheduler, controller

* Remove unneeded dependency from controller, scheduler

* Remove cluster creation from ShardingContainerPoolBalancer

* Remove trailing whitespace

Signed-off-by: Erika Hunhoff <[email protected]>

* Revert akka cluster changes to controller

Signed-off-by: Erika Hunhoff <[email protected]>

Signed-off-by: Erika Hunhoff <[email protected]>

Author: hunhoffe

Jenkinsfile

@@ -49,7 +49,7 @@ timeout(time: 12, unit: 'HOURS') {
                         sh "docker run -d --restart=always --name registry -v \"$HOME\"/certs:/certs \
                                 -e REGISTRY_HTTP_ADDR=0.0.0.0:${port} -e REGISTRY_HTTP_TLS_CERTIFICATE=/certs/${cert} \
                                 -e REGISTRY_HTTP_TLS_KEY=/certs/${key} -p ${port}:${port} registry:2"
-                        // Build the controller and invoker images.
+                        // Build the controller, scheduler, and invoker images.
                         sh "./gradlew distDocker -PdockerRegistry=${domainName}:${port}"
                         //Install the various modules like standalone
                         sh "./gradlew install"

build.gradle

@@ -42,7 +42,7 @@ subprojects {
         def cons = project.getDependencies().getConstraints()
         def akka = ['akka-actor', 'akka-cluster', 'akka-cluster-metrics', 'akka-cluster-tools', 'akka-coordination',
                     'akka-discovery', 'akka-distributed-data', 'akka-protobuf', 'akka-remote', 'akka-slf4j',
-                    'akka-stream', 'akka-stream-testkit', 'akka-testkit']
+                    'akka-stream', 'akka-stream-testkit', 'akka-testkit', 'akka-persistence', 'akka-cluster-sharding']
         def akkaHttp = ['akka-http', 'akka-http-core', 'akka-http-spray-json', 'akka-http-testkit', 'akka-http-xml',
                         'akka-parsing', 'akka-http2-support']
 

core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/ShardingContainerPoolBalancer.scala

@@ -158,7 +158,7 @@ class ShardingContainerPoolBalancer(
     AkkaManagement(actorSystem).start()
     ClusterBootstrap(actorSystem).start()
     Some(Cluster(actorSystem))
-  } else if (loadConfigOrThrow[Seq[String]]("akka.cluster.seed-nodes").nonEmpty) {
+  } else if (loadConfigOrThrow[Seq[String]]("akka.cluster.seed-nodes").nonEmpty) {  
     Some(Cluster(actorSystem))
   } else {
     None

core/scheduler/Dockerfile.cov

@@ -0,0 +1,30 @@
+#
+# 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.
+#
+
+FROM scheduler
+
+ARG OW_ROOT_DIR
+
+USER root
+RUN mkdir -p /coverage/common && \
+    mkdir -p /coverage/scheduler && \
+    mkdir -p "${OW_ROOT_DIR}/common/scala/build" && \
+    mkdir -p "${OW_ROOT_DIR}/core/scheduler/build" && \
+    ln -s /coverage/common "${OW_ROOT_DIR}/common/scala/build/scoverage" && \
+    ln -s /coverage/scheduler "${OW_ROOT_DIR}/core/scheduler/build/scoverage"
+
+COPY build/tmp/docker-coverage /scheduler/

core/scheduler/build.gradle

@@ -65,6 +65,8 @@ dependencies {
     implementation "org.scala-lang:scala-library:${gradle.scala.version}"
     implementation project(':common:scala')
     implementation "io.altoo:akka-kryo-serialization_${gradle.scala.depVersion}:1.0.0"
+    implementation "com.lightbend.akka.management:akka-management-cluster-bootstrap_${gradle.scala.depVersion}:${gradle.akka_management.version}"
+    implementation "com.lightbend.akka.discovery:akka-discovery-kubernetes-api_${gradle.scala.depVersion}:${gradle.akka_management.version}"
 }
 
 // workaround for akka-grpc

core/scheduler/src/main/resources/application.conf

@@ -44,9 +44,20 @@ akka {
       transport = tcp
     }
   }
+
+  cluster {
+    shutdown-after-unsuccessful-join-seed-nodes = 60s
+
+    # Disable legacy metrics in akka-cluster.
+    metrics.enabled = off
+  }
 }
 
 whisk {
+  cluster {
+    use-cluster-bootstrap: false
+  }
+
   # tracing configuration
   tracing {
     component = "Scheduler"

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/Scheduler.scala

@@ -21,6 +21,8 @@ import akka.Done
 import akka.actor.{ActorRef, ActorRefFactory, ActorSelection, ActorSystem, CoordinatedShutdown, Props}
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.management.scaladsl.AkkaManagement
+import akka.management.cluster.bootstrap.ClusterBootstrap
 import akka.pattern.ask
 import akka.util.Timeout
 import com.typesafe.config.ConfigValueFactory
@@ -258,6 +260,7 @@ trait SchedulerCore {
 object Scheduler {
 
   protected val protocol = loadConfigOrThrow[String]("whisk.scheduler.protocol")
+  protected val useClusterBootstrap = loadConfigOrThrow[Boolean]("whisk.cluster.use-cluster-bootstrap")
 
   val topicPrefix = loadConfigOrThrow[String](ConfigKeys.kafkaTopicsPrefix)
 
@@ -289,6 +292,11 @@ object Scheduler {
 
     implicit val logger = new AkkaLogging(akka.event.Logging.getLogger(actorSystem, this))
 
+    if (useClusterBootstrap) {
+      AkkaManagement(actorSystem).start()
+      ClusterBootstrap(actorSystem).start()
+    }
+
     // Prepare Kamon shutdown
     CoordinatedShutdown(actorSystem).addTask(CoordinatedShutdown.PhaseActorSystemTerminate, "shutdownKamon") { () =>
       logger.info(this, s"Shutting down Kamon with coordinated shutdown")

tools/admin/wskadmin

@@ -169,7 +169,7 @@ def parseArgs():
     subparser.required = True
 
     subcmd = subparser.add_parser('get', help='get logs')
-    subcmd.add_argument('components', help='components, one or more of [controllerN, invokerN] where N is the instance', nargs='*', default=['controller0', 'invoker0'])
+    subcmd.add_argument('components', help='components, one or more of [controllerN, schedulerN, invokerN] where N is the instance', nargs='*', default=['controller0', 'scheduler0', 'invoker0'])
     subcmd.add_argument('-t', '--tid', help='retrieve logs for the transaction id')
     subcmd.add_argument('-g', '--grep', help='retrieve logs that match grep expression')
 

tools/build/checkLogs.py

@@ -96,6 +96,7 @@ def colorize(code, string):
         ("db-triggers.log", {"db"}, [ partial(database_has_at_most_x_entries, 0) ]),
         # Assert that stdout of the container is correctly piped and empty
         ("controller0.log", {"system"}, [ partial(file_has_at_most_x_bytes, 0) ]),
+        ("scheduler0.log", {"system"}, [ partial(file_has_at_most_x_bytes, 0) ]),
         ("invoker0.log", {"system"}, [ partial(file_has_at_most_x_bytes, 0) ])
     ]
 

tools/build/citool

@@ -87,6 +87,7 @@ def parseArgs():
     subparser.add_argument('-s', '--sort', help='sort logs by timestamp', action='store_true')
     subparser.add_argument('-n', '--invokers', help='number of invokers', type=int, default=3)
     subparser.add_argument('-c', '--controllers', help='number of controllers', type=int, default=1)
+    subparser.add_argument('-c', '--schedulers', help='number of schedulers', type=int, default=1)
 
     return parser.parse_args()
 
@@ -312,6 +313,7 @@ def cat(args):
     else:
         components = {
             'controller': args.controllers,
+            'scheduler': args.schedulers,
             'invoker': args.invokers
         }
         logs = map(getComponentLogs, getComponentList(components))