| /* |
| * 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)) |
| } |
