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