Remove Mesos container factory. (#5133)
Remove Mesos container factory and related library, tests, and documentation
diff --git a/common/scala/build.gradle b/common/scala/build.gradle
index 1dddd97..d38e4f4 100644
--- a/common/scala/build.gradle
+++ b/common/scala/build.gradle
@@ -70,9 +70,6 @@
compile "io.kamon:kamon-prometheus_${gradle.scala.depVersion}:2.1.12"
compile "io.kamon:kamon-datadog_${gradle.scala.depVersion}:2.1.12"
- //for mesos
- compile "com.adobe.api.platform.runtime:mesos-actor:0.0.17"
-
// for etcd
compile("com.ibm.etcd:etcd-java:0.0.13")
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/mesos/MesosContainerFactory.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/mesos/MesosContainerFactory.scala
deleted file mode 100644
index b44d147..0000000
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/mesos/MesosContainerFactory.scala
+++ /dev/null
@@ -1,227 +0,0 @@
-/*
- * 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.mesos
-
-import akka.actor.ActorRef
-import akka.actor.ActorSystem
-import akka.pattern.ask
-import com.adobe.api.platform.runtime.mesos.Constraint
-import com.adobe.api.platform.runtime.mesos.LIKE
-import com.adobe.api.platform.runtime.mesos.LocalTaskStore
-import com.adobe.api.platform.runtime.mesos.MesosClient
-import com.adobe.api.platform.runtime.mesos.Subscribe
-import com.adobe.api.platform.runtime.mesos.SubscribeComplete
-import com.adobe.api.platform.runtime.mesos.Teardown
-import com.adobe.api.platform.runtime.mesos.UNLIKE
-import java.time.Instant
-
-import pureconfig._
-import pureconfig.generic.auto._
-
-import scala.concurrent.Await
-import scala.concurrent.ExecutionContext
-import scala.concurrent.Future
-import scala.concurrent.TimeoutException
-import scala.concurrent.duration._
-import scala.util.Try
-import org.apache.openwhisk.common.Counter
-import org.apache.openwhisk.common.Logging
-import org.apache.openwhisk.common.TransactionId
-import org.apache.openwhisk.core.ConfigKeys
-import org.apache.openwhisk.core.WhiskConfig
-import org.apache.openwhisk.core.containerpool._
-import org.apache.openwhisk.core.entity.ByteSize
-import org.apache.openwhisk.core.entity.ExecManifest
-import org.apache.openwhisk.core.entity.InvokerInstanceId
-import org.apache.openwhisk.core.entity.UUID
-
-/**
- * Configuration for mesos timeouts
- */
-case class MesosTimeoutConfig(failover: FiniteDuration,
- taskLaunch: FiniteDuration,
- taskDelete: FiniteDuration,
- subscribe: FiniteDuration,
- teardown: FiniteDuration)
-
-/**
- * Configuration for mesos action container health checks
- */
-case class MesosContainerHealthCheckConfig(portIndex: Int,
- delay: FiniteDuration,
- interval: FiniteDuration,
- timeout: FiniteDuration,
- gracePeriod: FiniteDuration,
- maxConsecutiveFailures: Int)
-
-/**
- * Configuration for MesosClient
- */
-case class MesosConfig(masterUrl: String,
- masterPublicUrl: Option[String],
- role: String,
- mesosLinkLogMessage: Boolean,
- constraints: Seq[String],
- constraintDelimiter: String,
- blackboxConstraints: Seq[String],
- teardownOnExit: Boolean,
- healthCheck: Option[MesosContainerHealthCheckConfig],
- offerRefuseDuration: FiniteDuration,
- heartbeatMaxFailures: Int,
- timeouts: MesosTimeoutConfig) {}
-
-class MesosContainerFactory(config: WhiskConfig,
- actorSystem: ActorSystem,
- logging: Logging,
- parameters: Map[String, Set[String]],
- containerArgs: ContainerArgsConfig =
- loadConfigOrThrow[ContainerArgsConfig](ConfigKeys.containerArgs),
- runtimesRegistryConfig: RuntimesRegistryConfig =
- loadConfigOrThrow[RuntimesRegistryConfig](ConfigKeys.runtimesRegistry),
- userImagesRegistryConfig: RuntimesRegistryConfig =
- loadConfigOrThrow[RuntimesRegistryConfig](ConfigKeys.userImagesRegistry),
- mesosConfig: MesosConfig = loadConfigOrThrow[MesosConfig](ConfigKeys.mesos),
- clientFactory: (ActorSystem, MesosConfig) => ActorRef = MesosContainerFactory.createClient,
- taskIdGenerator: () => String = MesosContainerFactory.taskIdGenerator _)
- extends ContainerFactory {
-
- implicit val as: ActorSystem = actorSystem
- implicit val ec: ExecutionContext = actorSystem.dispatcher
-
- /** Inits Mesos framework. */
- val mesosClientActor = clientFactory(as, mesosConfig)
-
- @volatile
- private var closed: Boolean = false
-
- subscribe()
-
- /** Subscribes Mesos actor to mesos event stream; retry on timeout (which should be unusual). */
- private def subscribe(): Future[Unit] = {
- logging.info(this, s"subscribing to Mesos master at ${mesosConfig.masterUrl}")
- mesosClientActor
- .ask(Subscribe)(mesosConfig.timeouts.subscribe)
- .mapTo[SubscribeComplete]
- .map(complete => logging.info(this, s"subscribe completed successfully... $complete"))
- .recoverWith {
- case e =>
- logging.error(this, s"subscribe failed... $e}")
- if (closed) Future.successful(()) else subscribe()
- }
- }
-
- override def createContainer(tid: TransactionId,
- name: String,
- actionImage: ExecManifest.ImageName,
- userProvidedImage: Boolean,
- memory: ByteSize,
- cpuShares: Int)(implicit config: WhiskConfig, logging: Logging): Future[Container] = {
- implicit val transid = tid
- val image = actionImage.resolveImageName(Some(
- ContainerFactory.resolveRegistryConfig(userProvidedImage, runtimesRegistryConfig, userImagesRegistryConfig).url))
- val constraintStrings = if (userProvidedImage) {
- mesosConfig.blackboxConstraints
- } else {
- mesosConfig.constraints
- }
-
- MesosTask.create(
- mesosClientActor,
- mesosConfig,
- taskIdGenerator,
- tid,
- image,
- userProvidedImage = userProvidedImage,
- memory = memory,
- cpuShares = cpuShares,
- environment = Map("__OW_API_HOST" -> config.wskApiHost) ++ containerArgs.extraEnvVarMap,
- network = containerArgs.network,
- dnsServers = containerArgs.dnsServers,
- name = Some(name),
- //strip any "--" prefixes on parameters (should make this consistent everywhere else)
- parameters
- .map({ case (k, v) => if (k.startsWith("--")) (k.replaceFirst("--", ""), v) else (k, v) })
- ++ containerArgs.extraArgs,
- parseConstraints(constraintStrings))
- }
-
- /**
- * Validate that constraint strings are well formed, and ignore constraints with unknown operators
- * @param constraintStrings
- * @param logging
- * @return
- */
- def parseConstraints(constraintStrings: Seq[String])(implicit logging: Logging): Seq[Constraint] =
- constraintStrings.flatMap(cs => {
- val parts = cs.split(mesosConfig.constraintDelimiter)
- require(parts.length == 3, "constraint must be in the form <attribute><delimiter><operator><delimiter><value>")
- Seq(LIKE, UNLIKE).find(_.toString == parts(1)) match {
- case Some(o) => Some(Constraint(parts(0), o, parts(2)))
- case _ =>
- logging.warn(this, s"ignoring unsupported constraint operator ${parts(1)}")
- None
- }
- })
-
- override def init(): Unit = ()
-
- /** Cleanups any remaining Containers; should block until complete; should ONLY be run at shutdown. */
- override def cleanup(): Unit = {
- val complete: Future[Any] = mesosClientActor.ask(Teardown)(mesosConfig.timeouts.teardown)
- Try(Await.result(complete, mesosConfig.timeouts.teardown))
- .map(_ => logging.info(this, "Mesos framework teardown completed."))
- .recover {
- case _: TimeoutException => logging.error(this, "Mesos framework teardown took too long.")
- case t: Throwable =>
- logging.error(this, s"Mesos framework teardown failed : $t}")
- }
- }
-
- def close(): Unit = {
- closed = true
- }
-}
-object MesosContainerFactory {
- private def createClient(actorSystem: ActorSystem, mesosConfig: MesosConfig): ActorRef =
- actorSystem.actorOf(
- MesosClient
- .props(
- () => "whisk-containerfactory-" + UUID(),
- "whisk-containerfactory-framework",
- mesosConfig.masterUrl,
- mesosConfig.role,
- mesosConfig.timeouts.failover,
- taskStore = new LocalTaskStore,
- refuseSeconds = mesosConfig.offerRefuseDuration.toSeconds.toDouble,
- heartbeatMaxFailures = mesosConfig.heartbeatMaxFailures))
-
- val counter = new Counter()
- val startTime = Instant.now.getEpochSecond
- private def taskIdGenerator(): String = {
- s"whisk-${counter.next()}-${startTime}"
- }
-}
-
-object MesosContainerFactoryProvider extends ContainerFactoryProvider {
- override def instance(actorSystem: ActorSystem,
- logging: Logging,
- config: WhiskConfig,
- instance: InvokerInstanceId,
- parameters: Map[String, Set[String]]): ContainerFactory =
- new MesosContainerFactory(config, actorSystem, logging, parameters)
-}
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/mesos/MesosTask.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/mesos/MesosTask.scala
deleted file mode 100644
index fc42b24..0000000
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/mesos/MesosTask.scala
+++ /dev/null
@@ -1,232 +0,0 @@
-/*
- * 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.mesos
-
-import akka.actor.ActorRef
-import akka.actor.ActorSystem
-import akka.event.Logging.ErrorLevel
-import akka.event.Logging.InfoLevel
-import akka.pattern.AskTimeoutException
-import akka.pattern.ask
-import akka.stream.scaladsl.Source
-import akka.util.ByteString
-import akka.util.Timeout
-import com.adobe.api.platform.runtime.mesos.Bridge
-import com.adobe.api.platform.runtime.mesos.CommandDef
-import com.adobe.api.platform.runtime.mesos.Constraint
-import com.adobe.api.platform.runtime.mesos.DeleteTask
-import com.adobe.api.platform.runtime.mesos.HealthCheckConfig
-import com.adobe.api.platform.runtime.mesos.Host
-import com.adobe.api.platform.runtime.mesos.Running
-import com.adobe.api.platform.runtime.mesos.SubmitTask
-import com.adobe.api.platform.runtime.mesos.TaskDef
-import com.adobe.api.platform.runtime.mesos.User
-import java.time.Instant
-import scala.concurrent.ExecutionContext
-import scala.concurrent.Future
-import scala.util.Failure
-import scala.util.Success
-import spray.json._
-import org.apache.openwhisk.common.Logging
-import org.apache.openwhisk.common.LoggingMarkers
-import org.apache.openwhisk.common.MetricEmitter
-import org.apache.openwhisk.common.TransactionId
-import org.apache.openwhisk.core.containerpool.Container
-import org.apache.openwhisk.core.containerpool.ContainerAddress
-import org.apache.openwhisk.core.containerpool.ContainerId
-import org.apache.openwhisk.core.containerpool.logging.LogLine
-import org.apache.openwhisk.core.entity.ByteSize
-import org.apache.openwhisk.core.entity.size._
-
-/**
- * MesosTask implementation of Container.
- * Differences from DockerContainer include:
- * - does not launch container using docker cli, but rather a Mesos framework
- * - does not support pause/resume
- * - does not support log collection (currently), but does provide a message indicating logs can be viewed via Mesos UI
- * (external log collection and retrieval must be enabled via LogStore SPI to expose logs to wsk cli)
- */
-case object Environment
-case class CreateContainer(image: String, memory: String, cpuShare: String)
-
-object MesosTask {
-
- val LAUNCH_CMD = "launch"
- val KILL_CMD = "kill"
-
- def create(mesosClientActor: ActorRef,
- mesosConfig: MesosConfig,
- taskIdGenerator: () => String,
- transid: TransactionId,
- image: String,
- userProvidedImage: Boolean = false,
- memory: ByteSize = 256.MB,
- cpuShares: Int = 0,
- environment: Map[String, String] = Map.empty,
- network: String = "bridge",
- dnsServers: Seq[String] = Seq.empty,
- name: Option[String] = None,
- parameters: Map[String, Set[String]] = Map.empty,
- constraints: Seq[Constraint] = Seq.empty)(implicit ec: ExecutionContext,
- log: Logging,
- as: ActorSystem): Future[Container] = {
- implicit val tid = transid
-
- val mesosCpuShares = cpuShares / 1024.0 // convert openwhisk (docker based) shares to mesos (cpu percentage)
- val mesosRam = memory.toMB.toInt
-
- val taskId = taskIdGenerator()
- val lowerNetwork = network.toLowerCase // match bridge+host without case, but retain case for user specified network
- val taskNetwork = lowerNetwork match {
- case "bridge" => Bridge
- case "host" => Host
- case _ => User(network)
- }
- val dnsOrEmpty = if (dnsServers.nonEmpty) Map("dns" -> dnsServers.toSet) else Map.empty[String, Set[String]]
-
- //transform our config to mesos-actor config:
- val healthCheckConfig = mesosConfig.healthCheck.map(
- c =>
- HealthCheckConfig(
- c.portIndex,
- c.delay.toSeconds.toDouble,
- c.interval.toSeconds.toDouble,
- c.timeout.toSeconds.toDouble,
- c.gracePeriod.toSeconds.toDouble,
- c.maxConsecutiveFailures))
- //define task
- val task = new TaskDef(
- taskId,
- name.getOrElse(image), // task name either the indicated name, or else the image name
- image,
- mesosCpuShares,
- mesosRam,
- List(8080), // all action containers listen on 8080
- healthCheckConfig, // port at index 0 used for health
- false,
- taskNetwork,
- dnsOrEmpty ++ parameters,
- Some(CommandDef(environment)),
- constraints.toSet)
-
- val taskLaunchTimeout = Timeout(mesosConfig.timeouts.taskLaunch)
- val start = transid.started(
- this,
- LoggingMarkers.INVOKER_MESOS_CMD(LAUNCH_CMD),
- s"launching mesos task for taskid $taskId (image:$image, mem: $mesosRam, cpu: $mesosCpuShares) (timeout: $taskLaunchTimeout)",
- logLevel = InfoLevel)
-
- val launched: Future[Running] =
- mesosClientActor.ask(SubmitTask(task))(taskLaunchTimeout).mapTo[Running]
-
- launched
- .andThen {
- case Success(taskDetails) =>
- transid.finished(this, start, s"launched task ${taskId} at ${taskDetails.hostname}:${taskDetails
- .hostports(0)}", logLevel = InfoLevel)
- case Failure(ate: AskTimeoutException) =>
- transid.failed(this, start, s"task launch timed out ${ate.getMessage}", ErrorLevel)
- MetricEmitter.emitCounterMetric(LoggingMarkers.INVOKER_MESOS_CMD_TIMEOUT(LAUNCH_CMD))
- //kill the task whose launch timed out
- destroy(mesosClientActor, mesosConfig, taskId)
- case Failure(t) =>
- //kill the task whose launch timed out
- destroy(mesosClientActor, mesosConfig, taskId)
- transid.failed(this, start, s"task launch failed ${t.getMessage}", ErrorLevel)
- }
- .map(taskDetails => {
- val taskHost = taskDetails.hostname
- val taskPort = taskDetails.hostports(0)
- val containerIp = new ContainerAddress(taskHost, taskPort)
- val containerId = new ContainerId(taskId);
- new MesosTask(containerId, containerIp, ec, log, as, taskId, mesosClientActor, mesosConfig)
- })
-
- }
- private def destroy(mesosClientActor: ActorRef, mesosConfig: MesosConfig, taskId: String)(
- implicit transid: TransactionId,
- logging: Logging,
- ec: ExecutionContext): Future[Unit] = {
- val taskDeleteTimeout = Timeout(mesosConfig.timeouts.taskDelete)
-
- val start = transid.started(
- this,
- LoggingMarkers.INVOKER_MESOS_CMD(MesosTask.KILL_CMD),
- s"killing mesos taskid $taskId (timeout: ${taskDeleteTimeout})",
- logLevel = InfoLevel)
-
- mesosClientActor
- .ask(DeleteTask(taskId))(taskDeleteTimeout)
- .andThen {
- case Success(_) => transid.finished(this, start, logLevel = InfoLevel)
- case Failure(ate: AskTimeoutException) =>
- transid.failed(this, start, s"task destroy timed out ${ate.getMessage}", ErrorLevel)
- MetricEmitter.emitCounterMetric(LoggingMarkers.INVOKER_MESOS_CMD_TIMEOUT(MesosTask.KILL_CMD))
- case Failure(t) => transid.failed(this, start, s"task destroy failed ${t.getMessage}", ErrorLevel)
- }
- .map(_ => {})
- }
-}
-
-object JsonFormatters extends DefaultJsonProtocol {
- implicit val createContainerJson = jsonFormat3(CreateContainer)
-}
-
-class MesosTask(override protected val id: ContainerId,
- override protected[core] val addr: ContainerAddress,
- override protected implicit val ec: ExecutionContext,
- override protected implicit val logging: Logging,
- override protected val as: ActorSystem,
- taskId: String,
- mesosClientActor: ActorRef,
- mesosConfig: MesosConfig)
- extends Container {
-
- /** Stops the container from consuming CPU cycles. */
- override def suspend()(implicit transid: TransactionId): Future[Unit] = {
- super.suspend()
- // suspend not supported (just return result from super)
- }
-
- /** Dual of halt. */
- override def resume()(implicit transid: TransactionId): Future[Unit] = {
- super.resume()
- // resume not supported (just return result from super)
- }
-
- /** Completely destroys this instance of the container. */
- override def destroy()(implicit transid: TransactionId): Future[Unit] = {
- MesosTask.destroy(mesosClientActor, mesosConfig, taskId)
- }
-
- /**
- * Obtains logs up to a given threshold from the container. Optionally waits for a sentinel to appear.
- * For Mesos, this log message is static per container, just indicating that Mesos logs can be found via the Mesos UI.
- * To disable this message, and just store an static log message per activation, set
- * org.apache.openwhisk.mesos.mesosLinkLogMessage=false
- */
- private val linkedLogMsg =
- s"Logs are not collected from Mesos containers currently. " +
- s"You can browse the logs for Mesos Task ID $taskId using the mesos UI at ${mesosConfig.masterPublicUrl
- .getOrElse(mesosConfig.masterUrl)}"
- private val noLinkLogMsg = "Log collection is not configured correctly, check with your service administrator."
- private val logMsg = if (mesosConfig.mesosLinkLogMessage) linkedLogMsg else noLinkLogMsg
- override def logs(limit: ByteSize, waitForSentinel: Boolean)(
- implicit transid: TransactionId): Source[ByteString, Any] =
- Source.single(ByteString(LogLine(logMsg, "stdout", Instant.now.toString).toJson.compactPrint))
-}
diff --git a/docs/mesos.md b/docs/mesos.md
deleted file mode 100644
index c8cd378..0000000
--- a/docs/mesos.md
+++ /dev/null
@@ -1,63 +0,0 @@
-<!--
-#
-# 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.
-#
--->
-# Mesos Support
-
-The `MesosContainerFactory` enables launching action containers within a Mesos cluster. It does not affect the deployment of OpenWhisk components (invoker, controller).
-
-## Enable
-
-To enable MesosContainerFactory, use the following TypeSafe Config properties
-
-| property | required | details | example |
-| --- | --- | --- | --- |
-| `whisk.spi.ContainerFactoryProvider` | required | enable the MesosContainerFactory | org.apache.openwhisk.core.mesos.MesosContainerFactoryProvider |
-| `whisk.mesos.master-url` | required | Mesos master HTTP endpoint to be accessed from the invoker for framework subscription | http://192.168.99.100:5050 |
-| `whisk.mesos.master-url-public` | optional (default to whisk.mesos.master-url) | public facing Mesos master HTTP endpoint for exposing logs to CLI users | http://192.168.99.100:5050 |
-| `whisk.mesos.role` | optional (default *) | Mesos framework role| any string e.g. `openwhisk` |
-| `whisk.mesos.failover-timeout-seconds` | optional (default 0) | how long to wait for the framework to reconnect with the same id before tasks are terminated | see http://mesos.apache.org/documentation/latest/high-availability-framework-guide/ |
-| `whisk.mesos.mesos-link-log-message` | optional (default true) | display a log message with a link to Mesos when using the default LogStore (or no log message) | Since logs are not available for invoker to collect from Mesos in general, you can either use an alternate LogStore or direct users to the Mesos UI | |
-| `whisk.mesos.constraints` | optional (default []) | placement constraint strings to use for managed containers | `["att1 LIKE v1", "att2 UNLIKE v2"]` | |
-| `whisk.mesos.blackbox-constraints` | optional (default []) | placement constraint strings to use for blackbox containers | `["att1 LIKE v1", "att2 UNLIKE v2"]` | |
-| `whisk.mesos.constraint-delimiter` | optional (default " ") | delimiter used to parse constraints | | |
-| `whisk.mesos.teardown-on-exit` | optional (default true) | set to true to disable the Mesos framework on system exit; set for false for HA deployments | | |
-
-To set these properties for your invoker, set the corresponding environment variables e.g.,
-```properties
-CONFIG_whisk_spi_ContainerFactoryProvider=org.apache.openwhisk.core.mesos.MesosContainerFactoryProvider
-CONFIG_whisk_mesos_masterUrl=http://192.168.99.100:5050
-```
-
-## Known Issues
-
-* Logs are not collected from action containers.
-
- For now, the Mesos public URL will be included in the logs retrieved via the wsk CLI. Once log retrieval from external sources is enabled, logs from Mesos containers would have to be routed to the external source, and then retrieved from that source.
-
-* No HA or failover support (single invoker per cluster).
-
- Currently the version of `mesos-actor` in use does not support HA or failover. Failover support is planned to be provided by:
-
- * multiple invokers running in an Akka cluster
- * the Mesos framework actor is a singleton within the cluster
- * the Mesos framework actor is available from the other invoker nodes
- * if the node that contains the Mesos framework actor fails:
- * the actor will be recreated on a separate invoker node
- * the actor will resubscribe to Mesos scheduler API with the same ID
- * the tasks that were previously launched by the actor will be reconciled
- * normal operation resumes
diff --git a/tests/src/test/scala/org/apache/openwhisk/core/containerpool/mesos/test/MesosContainerFactoryTest.scala b/tests/src/test/scala/org/apache/openwhisk/core/containerpool/mesos/test/MesosContainerFactoryTest.scala
deleted file mode 100644
index c27a298..0000000
--- a/tests/src/test/scala/org/apache/openwhisk/core/containerpool/mesos/test/MesosContainerFactoryTest.scala
+++ /dev/null
@@ -1,337 +0,0 @@
-/*
- * 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.containerpool.mesos.test
-
-import akka.actor.ActorSystem
-import akka.stream.scaladsl.Sink
-import akka.testkit.TestKit
-import akka.testkit.TestProbe
-import com.adobe.api.platform.runtime.mesos.Bridge
-import com.adobe.api.platform.runtime.mesos.CommandDef
-import com.adobe.api.platform.runtime.mesos.Constraint
-import com.adobe.api.platform.runtime.mesos.DeleteTask
-import com.adobe.api.platform.runtime.mesos.LIKE
-import com.adobe.api.platform.runtime.mesos.Running
-import com.adobe.api.platform.runtime.mesos.SubmitTask
-import com.adobe.api.platform.runtime.mesos.Subscribe
-import com.adobe.api.platform.runtime.mesos.SubscribeComplete
-import com.adobe.api.platform.runtime.mesos.TaskDef
-import com.adobe.api.platform.runtime.mesos.UNLIKE
-import com.adobe.api.platform.runtime.mesos.User
-import common.StreamLogging
-import org.apache.mesos.v1.Protos.TaskID
-import org.apache.mesos.v1.Protos.TaskState
-import org.apache.mesos.v1.Protos.TaskStatus
-import org.junit.runner.RunWith
-import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, FlatSpecLike, Matchers}
-import org.scalatest.junit.JUnitRunner
-
-import scala.collection.immutable.Map
-import scala.concurrent.Await
-import scala.concurrent.Future
-import scala.concurrent.duration._
-import org.apache.openwhisk.common.TransactionId
-import org.apache.openwhisk.core.WhiskConfig
-import org.apache.openwhisk.core.WhiskConfig._
-import org.apache.openwhisk.core.containerpool.ContainerArgsConfig
-import org.apache.openwhisk.core.containerpool.ContainerPoolConfig
-import org.apache.openwhisk.core.containerpool.logging.DockerToActivationLogStore
-import org.apache.openwhisk.core.entity.ExecManifest.ImageName
-import org.apache.openwhisk.core.entity.size._
-import org.apache.openwhisk.core.mesos.MesosConfig
-import org.apache.openwhisk.core.mesos.MesosContainerFactory
-import org.apache.openwhisk.core.mesos.MesosTimeoutConfig
-import org.apache.openwhisk.utils.retry
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[JUnitRunner])
-class MesosContainerFactoryTest
- extends TestKit(ActorSystem("MesosActorSystem"))
- with FlatSpecLike
- with Matchers
- with StreamLogging
- with BeforeAndAfterEach
- with BeforeAndAfterAll {
-
- /** Awaits the given future, throws the exception enclosed in Failure. */
- def await[A](f: Future[A], timeout: FiniteDuration = 500.milliseconds) = Await.result[A](f, timeout)
-
- implicit val wskConfig =
- new WhiskConfig(Map(wskApiHostname -> "apihost") ++ wskApiHost)
- var count = 0
- var lastTaskId: String = null
- def testTaskId() = {
- count += 1
- lastTaskId = "testTaskId" + count
- lastTaskId
- }
-
- // 80 slots, each 265MB
- val poolConfig = ContainerPoolConfig(21200.MB, 0.5, false, 2.second, 1.minute, None, 100, 3, false, 1.seconds)
- val actionMemory = 265.MB
- val mesosCpus = poolConfig.cpuShare(actionMemory) / 1024.0
-
- val containerArgsConfig =
- new ContainerArgsConfig(
- "net1",
- Seq("dns1", "dns2"),
- Seq.empty,
- Seq.empty,
- Seq.empty,
- Map("extra1" -> Set("e1", "e2"), "extra2" -> Set("e3", "e4")))
-
- private var factory: MesosContainerFactory = _
- override def beforeEach() = {
- stream.reset()
- }
-
- override protected def afterEach(): Unit = {
- super.afterEach()
- Option(factory).foreach(_.close())
- }
-
- override def afterAll(): Unit = {
- TestKit.shutdownActorSystem(system, verifySystemShutdown = true)
- retry({
- val threadNames = Thread.getAllStackTraces.asScala.keySet.map(_.getName)
- withClue(s"Threads related to MesosActorSystem found to be active $threadNames") {
- assert(!threadNames.exists(_.startsWith("MesosActorSystem")))
- }
- }, 10, Some(1.second))
- super.afterAll()
- }
-
- val timeouts = MesosTimeoutConfig(1.seconds, 1.seconds, 1.seconds, 1.seconds, 1.seconds)
-
- val mesosConfig =
- MesosConfig("http://master:5050", None, "*", true, Seq.empty, " ", Seq.empty, true, None, 1.seconds, 2, timeouts)
-
- behavior of "MesosContainerFactory"
-
- it should "send Subscribe on init" in {
- val wskConfig = new WhiskConfig(Map.empty)
- factory = new MesosContainerFactory(
- wskConfig,
- system,
- logging,
- Map("--arg1" -> Set("v1", "v2")),
- containerArgsConfig,
- mesosConfig = mesosConfig,
- clientFactory = (system, mesosConfig) => testActor)
-
- expectMsg(Subscribe)
- }
-
- it should "send SubmitTask (with constraints) on create" in {
- val mesosConfig = MesosConfig(
- "http://master:5050",
- None,
- "*",
- true,
- Seq("att1 LIKE v1", "att2 UNLIKE v2"),
- " ",
- Seq("bbatt1 LIKE v1", "bbatt2 UNLIKE v2"),
- true,
- None,
- 1.seconds,
- 2,
- timeouts)
-
- factory = new MesosContainerFactory(
- wskConfig,
- system,
- logging,
- Map("--arg1" -> Set("v1", "v2"), "--arg2" -> Set("v3", "v4"), "other" -> Set("v5", "v6")),
- containerArgsConfig,
- mesosConfig = mesosConfig,
- clientFactory = (_, _) => testActor,
- taskIdGenerator = testTaskId _)
-
- expectMsg(Subscribe)
- factory.createContainer(
- TransactionId.testing,
- "mesosContainer",
- ImageName("fakeImage"),
- false,
- actionMemory,
- poolConfig.cpuShare(actionMemory))
-
- expectMsg(
- SubmitTask(TaskDef(
- lastTaskId,
- "mesosContainer",
- "fakeImage",
- mesosCpus,
- actionMemory.toMB.toInt,
- List(8080),
- None,
- false,
- User("net1"),
- Map(
- "arg1" -> Set("v1", "v2"),
- "arg2" -> Set("v3", "v4"),
- "other" -> Set("v5", "v6"),
- "dns" -> Set("dns1", "dns2"),
- "extra1" -> Set("e1", "e2"),
- "extra2" -> Set("e3", "e4")),
- Some(CommandDef(Map("__OW_API_HOST" -> wskConfig.wskApiHost))),
- Seq(Constraint("att1", LIKE, "v1"), Constraint("att2", UNLIKE, "v2")).toSet)))
- }
-
- it should "send DeleteTask on destroy" in {
- val probe = TestProbe()
- factory = new MesosContainerFactory(
- wskConfig,
- system,
- logging,
- Map("--arg1" -> Set("v1", "v2"), "--arg2" -> Set("v3", "v4"), "other" -> Set("v5", "v6")),
- containerArgsConfig,
- mesosConfig = mesosConfig,
- clientFactory = (system, mesosConfig) => probe.testActor,
- taskIdGenerator = testTaskId _)
-
- probe.expectMsg(Subscribe)
- //emulate successful subscribe
- probe.reply(new SubscribeComplete("testid"))
-
- //create the container
- val c = factory.createContainer(
- TransactionId.testing,
- "mesosContainer",
- ImageName("fakeImage"),
- false,
- actionMemory,
- poolConfig.cpuShare(actionMemory))
- probe.expectMsg(
- SubmitTask(TaskDef(
- lastTaskId,
- "mesosContainer",
- "fakeImage",
- mesosCpus,
- actionMemory.toMB.toInt,
- List(8080),
- None,
- false,
- User("net1"),
- Map(
- "arg1" -> Set("v1", "v2"),
- "arg2" -> Set("v3", "v4"),
- "other" -> Set("v5", "v6"),
- "dns" -> Set("dns1", "dns2"),
- "extra1" -> Set("e1", "e2"),
- "extra2" -> Set("e3", "e4")),
- Some(CommandDef(Map("__OW_API_HOST" -> wskConfig.wskApiHost))))))
-
- //emulate successful task launch
- val taskId = TaskID.newBuilder().setValue(lastTaskId)
-
- probe.reply(
- Running(
- taskId.getValue,
- "testAgentID",
- TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_RUNNING).build(),
- "agenthost",
- Seq(30000)))
- //wait for container
- val container = await(c)
-
- //destroy the container
- implicit val tid = TransactionId.testing
- val deleted = container.destroy()
- probe.expectMsg(DeleteTask(lastTaskId))
-
- probe.reply(TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_KILLED).build())
-
- await(deleted)
-
- }
-
- it should "return static message for logs" in {
- val probe = TestProbe()
- factory = new MesosContainerFactory(
- wskConfig,
- system,
- logging,
- Map("--arg1" -> Set("v1", "v2"), "--arg2" -> Set("v3", "v4"), "other" -> Set("v5", "v6")),
- new ContainerArgsConfig(
- "bridge",
- Seq.empty,
- Seq.empty,
- Seq.empty,
- Seq.empty,
- Map("extra1" -> Set("e1", "e2"), "extra2" -> Set("e3", "e4"))),
- mesosConfig = mesosConfig,
- clientFactory = (system, mesosConfig) => probe.testActor,
- taskIdGenerator = testTaskId _)
-
- probe.expectMsg(Subscribe)
- //emulate successful subscribe
- probe.reply(new SubscribeComplete("testid"))
-
- //create the container
- val c = factory.createContainer(
- TransactionId.testing,
- "mesosContainer",
- ImageName("fakeImage"),
- false,
- actionMemory,
- poolConfig.cpuShare(actionMemory))
-
- probe.expectMsg(
- SubmitTask(TaskDef(
- lastTaskId,
- "mesosContainer",
- "fakeImage",
- mesosCpus,
- actionMemory.toMB.toInt,
- List(8080),
- None,
- false,
- Bridge,
- Map(
- "arg1" -> Set("v1", "v2"),
- "arg2" -> Set("v3", "v4"),
- "other" -> Set("v5", "v6"),
- "extra1" -> Set("e1", "e2"),
- "extra2" -> Set("e3", "e4")),
- Some(CommandDef(Map("__OW_API_HOST" -> wskConfig.wskApiHost))))))
-
- //emulate successful task launch
- val taskId = TaskID.newBuilder().setValue(lastTaskId)
-
- probe.reply(
- Running(
- taskId.getValue,
- "testAgentID",
- TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_RUNNING).build(),
- "agenthost",
- Seq(30000)))
- //wait for container
- val container = await(c)
-
- implicit val tid = TransactionId.testing
- val logs = container
- .logs(actionMemory, false)
- .via(DockerToActivationLogStore.toFormattedString)
- .runWith(Sink.seq)
- await(logs)(0) should endWith
- " stdout: Logs are not collected from Mesos containers currently. You can browse the logs for Mesos Task ID testTaskId using the mesos UI at http://master:5050"
-
- }
-}
