| ///* |
| // * 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.amaterasu.leader.mesos.schedulers |
| // |
| //import java.io.{File, PrintWriter, StringWriter} |
| //import java.nio.file.{Files, Path, Paths, StandardCopyOption} |
| //import java.util |
| //import java.util.{Collections, UUID} |
| //import java.util.concurrent.locks.ReentrantLock |
| //import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue} |
| // |
| //import com.fasterxml.jackson.databind.ObjectMapper |
| //import com.fasterxml.jackson.dataformat.yaml.YAMLFactory |
| //import com.fasterxml.jackson.module.scala.DefaultScalaModule |
| //import org.apache.amaterasu.common.configuration.{ClusterConfig, ConfigManager} |
| //import org.apache.amaterasu.common.configuration.enums.ActionStatus |
| //import org.apache.amaterasu.common.dataobjects.ActionData |
| //import org.apache.amaterasu.common.execution.actions.Notification |
| //import org.apache.amaterasu.common.execution.actions.enums.{NotificationLevel, NotificationType} |
| //import org.apache.amaterasu.leader.common.execution.{JobLoader, JobManager} |
| //import org.apache.amaterasu.leader.common.execution.frameworks.FrameworkProvidersFactory |
| //import org.apache.amaterasu.leader.common.utilities.DataLoader |
| //import org.apache.amaterasu.leader.utilities.HttpServer |
| //import org.apache.commons.io.FileUtils |
| //import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory} |
| //import org.apache.curator.retry.ExponentialBackoffRetry |
| //import org.apache.log4j.LogManager |
| //import org.apache.mesos.Protos.CommandInfo.URI |
| //import org.apache.mesos.Protos.Environment.Variable |
| //import org.apache.mesos.Protos._ |
| //import org.apache.mesos.protobuf.ByteString |
| //import org.apache.mesos.{Protos, SchedulerDriver} |
| // |
| //import scala.collection.JavaConverters._ |
| //import scala.collection.concurrent |
| //import scala.collection.concurrent.TrieMap |
| // |
| ///** |
| // * The JobScheduler is a Mesos implementation. It is in charge of scheduling the execution of |
| // * Amaterasu actions for a specific job |
| // */ |
| //class JobScheduler extends AmaterasuScheduler { |
| // |
| // /*private val props: Properties = new Properties(new File("")) |
| // private val version = props.getProperty("version") |
| // println(s"===> version $version")*/ |
| // LogManager.resetConfiguration() |
| // private var frameworkFactory: FrameworkProvidersFactory = _ |
| // private var configManager: ConfigManager = _ |
| // private var jobManager: JobManager = _ |
| // private var client: CuratorFramework = _ |
| // private var config: ClusterConfig = _ |
| // private var src: String = _ |
| // private var env: String = _ |
| // private var branch: String = _ |
| // private var userName: String = _ |
| // private var password: String = _ |
| // private var resume: Boolean = false |
| // private var reportLevel: NotificationLevel = _ |
| // |
| // private val jarFile = new File(this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath) |
| // private val amaDist = new File(s"${new File(jarFile.getParent).getParent}/dist") |
| // |
| // val slavesExecutors = new TrieMap[String, ExecutorInfo] |
| // private var awsEnv: String = "" |
| // |
| // // this map holds the following structure: |
| // // slaveId |
| // // | |
| // // +-> taskId, actionStatus) |
| // private val executionMap: concurrent.Map[String, concurrent.Map[String, ActionStatus]] = new ConcurrentHashMap[String, concurrent.Map[String, ActionStatus]].asScala |
| // private val lock = new ReentrantLock() |
| // private val offersToTaskIds: concurrent.Map[String, String] = new ConcurrentHashMap[String, String].asScala |
| // private val taskIdsToActions: concurrent.Map[Protos.TaskID, String] = new ConcurrentHashMap[Protos.TaskID, String].asScala |
| // |
| // private val mapper = new ObjectMapper() |
| // mapper.registerModule(DefaultScalaModule) |
| // |
| // private val yamlMapper = new ObjectMapper(new YAMLFactory()) |
| // yamlMapper.registerModule(DefaultScalaModule) |
| // |
| // def error(driver: SchedulerDriver, message: String): Unit = { |
| // log.error(s"===> $message") |
| // } |
| // |
| // def executorLost(driver: SchedulerDriver, executorId: ExecutorID, slaveId: SlaveID, status: Int) {} |
| // |
| // def slaveLost(driver: SchedulerDriver, slaveId: SlaveID) {} |
| // |
| // def disconnected(driver: SchedulerDriver) {} |
| // |
| // def frameworkMessage(driver: SchedulerDriver, executorId: ExecutorID, slaveId: SlaveID, data: Array[Byte]): Unit = { |
| // |
| // val notification = mapper.readValue(data, classOf[Notification]) |
| // |
| // reportLevel match { |
| // case NotificationLevel.Code => printNotification(notification) |
| // case NotificationLevel.Execution => |
| // if (notification.getNotLevel != NotificationLevel.Code) |
| // printNotification(notification) |
| // case _ => |
| // } |
| // |
| // } |
| // |
| // def statusUpdate(driver: SchedulerDriver, status: TaskStatus): Unit = { |
| // |
| // val actionName = taskIdsToActions(status.getTaskId) |
| // status.getState match { |
| // case TaskState.TASK_STARTING => log.info("Task starting ...") |
| // case TaskState.TASK_RUNNING => { |
| // jobManager.actionStarted(status.getTaskId.getValue) |
| // printNotification(new Notification("", s"created container for $actionName created", NotificationType.Info, NotificationLevel.Execution)) |
| // |
| // } |
| // case TaskState.TASK_FINISHED => { |
| // jobManager.actionComplete(status.getTaskId.getValue) |
| // printNotification(new Notification("", s"Container ${status.getExecutorId.getValue} Complete with task ${status.getTaskId.getValue} with success.", NotificationType.Info, NotificationLevel.Execution)) |
| // } |
| // case TaskState.TASK_FAILED | |
| // TaskState.TASK_KILLED | |
| // TaskState.TASK_ERROR | |
| // TaskState.TASK_LOST => { |
| // jobManager.actionFailed(status.getTaskId.getValue, status.getMessage) |
| // printNotification(new Notification("", s"error launching container with ${status.getMessage} in ${status.getData.toStringUtf8}", NotificationType.Error, NotificationLevel.Execution)) |
| // |
| // } |
| // case _ => log.warn("WTF? just got unexpected task state: " + status.getState) |
| // } |
| // |
| // } |
| // |
| // def validateOffer(offer: Offer): Boolean = { |
| // |
| // val resources = offer.getResourcesList.asScala |
| // |
| // resources.count(r => r.getName == "cpus" && r.getScalar.getValue >= config.jobs.tasks.cpus) > 0 && |
| // resources.count(r => r.getName == "mem" && r.getScalar.getValue >= config.jobs.tasks.mem) > 0 |
| // } |
| // |
| // def offerRescinded(driver: SchedulerDriver, offerId: OfferID): Unit = { |
| // |
| // val actionId = offersToTaskIds(offerId.getValue) |
| // jobManager.reQueueAction(actionId) |
| // |
| // } |
| // |
| // def resourceOffers(driver: SchedulerDriver, offers: util.List[Offer]): Unit = { |
| // |
| // println(jobManager.toString) |
| // |
| // for (offer <- offers.asScala) { |
| // |
| // if (validateOffer(offer)) { |
| // |
| // log.info(s"Accepting offer, id=${offer.getId}") |
| // |
| // // this is done to avoid the processing the same action |
| // // multiple times |
| // lock.lock() |
| // |
| // try { |
| // val actionData = jobManager.getNextActionData |
| // if (actionData != null) { |
| // |
| // frameworkFactory = FrameworkProvidersFactory(env, config) |
| // val items = frameworkFactory.providers.values.flatMap(_.getConfigurationItems).toList.asJava |
| // configManager = new ConfigManager(env, "repo", items) |
| // |
| // val taskId = Protos.TaskID.newBuilder().setValue(actionData.getId).build() |
| // taskIdsToActions.put(taskId, actionData.getName) |
| // // setting up the configuration files for the container |
| // val envYaml = configManager.getActionConfigContent(actionData.getName, actionData.getConfig) |
| // writeConfigFile(envYaml, jobManager.getJobId, actionData.getName, "env.yaml") |
| // |
| // val envConf = configManager.getActionConfiguration(actionData.getName, actionData.getConfig) |
| // val dataStores = DataLoader.getTaskData(actionData, env).getExports |
| // val writer = new StringWriter() |
| // yamlMapper.writeValue(writer, dataStores) |
| // val dataStoresYaml = writer.toString |
| // writeConfigFile(dataStoresYaml, jobManager.getJobId, actionData.getName, "datastores.yaml") |
| // |
| // writeConfigFile(s"jobId: ${jobManager.getJobId}\nactionName: ${actionData.getName}", jobManager.getJobId, actionData.getName, "runtime.yaml") |
| // |
| // val datasets = DataLoader.getDatasets(env) |
| // writeConfigFile(datasets, jobManager.getJobId, actionData.getName, "datasets.yaml") |
| // offersToTaskIds.put(offer.getId.getValue, taskId.getValue) |
| // |
| // // atomically adding a record for the slave, I'm storing all the actions |
| // // on a slave level to efficiently handle slave loses |
| // executionMap.putIfAbsent(offer.getSlaveId.toString, new ConcurrentHashMap[String, ActionStatus].asScala) |
| // |
| // val slaveActions = executionMap(offer.getSlaveId.toString) |
| // slaveActions.put(taskId.getValue, ActionStatus.Started) |
| // |
| // val frameworkProvider = frameworkFactory.providers(actionData.getGroupId) |
| // |
| // val runnerProvider = frameworkProvider.getRunnerProvider(actionData.getTypeId) |
| // |
| // printNotification(new Notification("", s"provider ${runnerProvider.getClass.getName}", NotificationType.Info, NotificationLevel.Execution)) |
| // // searching for an executor that already exist on the slave, if non exist |
| // // we create a new one |
| // var executor: ExecutorInfo = null |
| // |
| // // val slaveId = offer.getSlaveId.getValue |
| // // slavesExecutors.synchronized { |
| // |
| // val execData = DataLoader.getExecutorDataBytes(env, config) |
| // val executorId = taskId.getValue + "-" + UUID.randomUUID() |
| // //creating the command |
| // |
| // // // TODO: move this into the runner provider somehow |
| // // if(!actionData.getSrc.isEmpty){ |
| // // copy(get(s"repo/src/${actionData.getSrc}"), get(s"dist/${jobManager.getJobId}/${actionData.getName}/${actionData.getSrc}"), REPLACE_EXISTING) |
| // // } |
| // val commandStr = runnerProvider.getCommand(jobManager.getJobId, actionData, envConf, executorId, "") |
| // printNotification(new Notification("", s"container command $commandStr", NotificationType.Info, NotificationLevel.Execution)) |
| // |
| // val command = CommandInfo |
| // .newBuilder |
| // .setValue(commandStr) |
| // // .addUris(URI.newBuilder |
| // // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/executor-${config.version}-all.jar") |
| // // .setExecutable(false) |
| // // .setExtract(false) |
| // // .build()) |
| // |
| // // Getting framework (group) resources |
| // log.info(s"===> groupResources: ${frameworkProvider.getGroupResources}") |
| // frameworkProvider.getGroupResources.foreach(f => command.addUris(URI.newBuilder |
| // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/${f.getName}") |
| // .setExecutable(false) |
| // .setExtract(true) |
| // .build() |
| // )) |
| // |
| // // Getting runner resources |
| // runnerProvider.getRunnerResources.foreach(r => { |
| // command.addUris(URI.newBuilder |
| // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$r") |
| // .setExecutable(false) |
| // .setExtract(false) |
| // .build()) |
| // }) |
| // |
| // // Getting action dependencies |
| // runnerProvider.getActionDependencies(jobManager.getJobId, actionData).foreach(r => { |
| // |
| // FileUtils.copyFile(new File(r), new File(s"dist/$r")) |
| // command.addUris(URI.newBuilder |
| // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$r") |
| // .setExecutable(false) |
| // .setExtract(false) |
| // .build()) |
| // }) |
| // |
| // // Getting action specific resources |
| // runnerProvider.getActionResources(jobManager.getJobId, actionData).foreach(r => command.addUris(URI.newBuilder |
| // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$r") |
| // .setExecutable(false) |
| // .setExtract(false) |
| // .build())) |
| // |
| // // setting up action executable |
| // val sourcePath = new File(runnerProvider.getActionExecutable(jobManager.getJobId, actionData)) |
| // var executable: Path = null |
| // if (actionData.getHasArtifact) { |
| // val relativePath = amaDist.toPath.getRoot.relativize(sourcePath.toPath) |
| // executable = relativePath.subpath(amaDist.toPath.getNameCount, relativePath.getNameCount) |
| // } else { |
| // val dest = new File(s"dist/${jobManager.getJobId}/${sourcePath.toString}") |
| // FileUtils.copyFile(sourcePath, dest) |
| // executable = Paths.get(jobManager.getJobId, sourcePath.toPath.toString) |
| // } |
| // |
| // println(s"===> executable $executable") |
| // command.addUris(URI.newBuilder |
| // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/$executable") |
| // .setExecutable(false) |
| // .setExtract(false) |
| // .build()) |
| // |
| // command |
| // .addUris(URI.newBuilder() |
| // .setValue(s"http://${sys.env("AMA_NODE")}:${config.webserver.Port}/amaterasu.properties") |
| // .setExecutable(false) |
| // .setExtract(false) |
| // .build()) |
| // |
| // // setting the processes environment variables |
| // val envVarsList = frameworkProvider.getEnvironmentVariables.asScala.toList.map(x => Variable.newBuilder().setName(x._1).setValue(x._2).build()).asJava |
| // command.setEnvironment(Environment.newBuilder().addAllVariables(envVarsList)) |
| // |
| // executor = ExecutorInfo |
| // .newBuilder |
| // .setData(ByteString.copyFrom(execData)) |
| // .setName(taskId.getValue) |
| // .setExecutorId(ExecutorID.newBuilder().setValue(executorId)) |
| // .setCommand(command) |
| // .build() |
| // |
| // slavesExecutors.put(offer.getSlaveId.getValue, executor) |
| // |
| // |
| // val driverConfiguration = frameworkProvider.getDriverConfiguration(configManager) |
| // |
| // var actionTask: TaskInfo = null |
| // |
| // if (runnerProvider.getHasExecutor) { |
| // actionTask = TaskInfo |
| // .newBuilder |
| // .setName(taskId.getValue) |
| // .setTaskId(taskId) |
| // .setExecutor(executor) |
| // |
| // .setData(ByteString.copyFrom(DataLoader.getTaskDataBytes(actionData, env))) |
| // .addResources(createScalarResource("cpus", driverConfiguration.getCpus)) |
| // .addResources(createScalarResource("mem", driverConfiguration.getMemory)) |
| // .addResources(createScalarResource("disk", config.jobs.repoSize)) |
| // .setSlaveId(offer.getSlaveId) |
| // .build() |
| // |
| // //driver.launchTasks(Collections.singleton(offer.getId), List(actionTask).asJava) |
| // } |
| // else { |
| // actionTask = TaskInfo |
| // .newBuilder |
| // .setName(taskId.getValue) |
| // .setTaskId(taskId) |
| // .setCommand(command) |
| // |
| // //.setData(ByteString.copyFrom(DataLoader.getTaskDataBytes(actionData, env))) |
| // .addResources(createScalarResource("cpus", driverConfiguration.getCpus)) |
| // .addResources(createScalarResource("mem", driverConfiguration.getMemory)) |
| // .addResources(createScalarResource("disk", config.jobs.repoSize)) |
| // .setSlaveId(offer.getSlaveId) |
| // .build() |
| // |
| // //driver.launchTasks(Collections.singleton(offer.getId), List(actionTask).asJava) |
| // } |
| // |
| // printNotification(new Notification("", s"requesting container for ${actionData.getName}", NotificationType.Info, NotificationLevel.Execution)) |
| // driver.launchTasks(Collections.singleton(offer.getId), List(actionTask).asJava) |
| // |
| // } |
| // else if (jobManager.getOutOfActions) { |
| // log.info(s"framework ${jobManager.getJobId} execution finished") |
| // |
| // val repo = new File("repo/") |
| // repo.delete() |
| // |
| // HttpServer.stop() |
| // driver.declineOffer(offer.getId) |
| // driver.stop() |
| // sys.exit() |
| // } |
| // else { |
| // log.info("Declining offer, no action ready for execution") |
| // driver.declineOffer(offer.getId) |
| // } |
| // } |
| // finally { |
| // lock.unlock() |
| // } |
| // } |
| // else { |
| // log.info("Declining offer, no sufficient resources") |
| // driver.declineOffer(offer.getId) |
| // } |
| // |
| // } |
| // |
| // } |
| // |
| // def registered(driver: SchedulerDriver, frameworkId: FrameworkID, masterInfo: MasterInfo): Unit = { |
| // |
| // if (!resume) { |
| // |
| // jobManager = JobLoader.loadJob( |
| // src, |
| // branch, |
| // frameworkId.getValue, |
| // userName, |
| // password, |
| // client, |
| // config.jobs.tasks.attempts, |
| // new LinkedBlockingQueue[ActionData]() |
| // ) |
| // } |
| // else { |
| // |
| // JobLoader.reloadJob( |
| // frameworkId.getValue, |
| // userName, |
| // password, |
| // client, |
| // config.jobs.tasks.attempts, |
| // new LinkedBlockingQueue[ActionData]() |
| // ) |
| // |
| // } |
| // |
| // |
| // |
| // |
| // jobManager.start() |
| // |
| // createJobDir(jobManager.getJobId) |
| // |
| // } |
| // |
| // def reregistered(driver: SchedulerDriver, masterInfo: Protos.MasterInfo) {} |
| // |
| // def printNotification(notification: Notification): Unit = { |
| // |
| // var color = Console.WHITE |
| // |
| // notification.getNotType match { |
| // |
| // case NotificationType.Info => |
| // color = Console.WHITE |
| // println(s"$color${Console.BOLD}===> ${notification.getMsg} ${Console.RESET}") |
| // case NotificationType.Success => |
| // color = Console.GREEN |
| // println(s"$color${Console.BOLD}===> ${notification.getLine} ${Console.RESET}") |
| // case NotificationType.Error => |
| // color = Console.RED |
| // println(s"$color${Console.BOLD}===> ${notification.getLine} ${Console.RESET}") |
| // println(s"$color${Console.BOLD}===> ${notification.getMsg} ${Console.RESET}") |
| // |
| // } |
| // |
| // } |
| // |
| // private def createJobDir(jobId: String): Unit = { |
| // val jarFile = new File(this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath) |
| // val amaHome = new File(jarFile.getParent).getParent |
| // val jobDir = s"$amaHome/dist/$jobId/" |
| // |
| // val dir = new File(jobDir) |
| // if (!dir.exists()) { |
| // dir.mkdir() |
| // } |
| // } |
| // |
| // /** |
| // * This function creates an action specific env.yml file int the dist folder with the following path: |
| // * dist/{jobId}/{actionName}/env.yml to be added to the container |
| // * |
| // * @param configuration A YAML string to be written to the env file |
| // * @param jobId the jobId |
| // * @param actionName the name of the action |
| // */ |
| // def writeConfigFile(configuration: String, jobId: String, actionName: String, fileName: String): Unit = { |
| // val jarFile = new File(this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath) |
| // val amaHome = new File(jarFile.getParent).getParent |
| // val envLocation = s"$amaHome/dist/$jobId/$actionName/" |
| // |
| // val dir = new File(envLocation) |
| // if (!dir.exists()) { |
| // dir.mkdirs() |
| // } |
| // |
| // new PrintWriter(s"$envLocation/$fileName") { |
| // write(configuration) |
| // close |
| // } |
| // } |
| //} |
| // |
| //object JobScheduler { |
| // |
| // def apply(src: String, |
| // branch: String, |
| // username: String, |
| // password: String, |
| // env: String, |
| // resume: Boolean, |
| // config: ClusterConfig, |
| // report: String, |
| // home: String): JobScheduler = { |
| // |
| // LogManager.resetConfiguration() |
| // val scheduler = new JobScheduler() |
| // |
| // HttpServer.start(config.webserver.Port, s"$home/${config.webserver.Root}") |
| // |
| // if (sys.env.get("AWS_ACCESS_KEY_ID").isDefined && |
| // sys.env.get("AWS_SECRET_ACCESS_KEY").isDefined) { |
| // |
| // scheduler.awsEnv = s"env AWS_ACCESS_KEY_ID=${sys.env("AWS_ACCESS_KEY_ID")} env AWS_SECRET_ACCESS_KEY=${sys.env("AWS_SECRET_ACCESS_KEY")}" |
| // } |
| // |
| // scheduler.resume = resume |
| // scheduler.src = src |
| // scheduler.branch = branch |
| // scheduler.userName = username |
| // scheduler.password = password |
| // scheduler.env = env |
| // scheduler.reportLevel = NotificationLevel.valueOf(report.capitalize) |
| // |
| // val retryPolicy = new ExponentialBackoffRetry(1000, 3) |
| // scheduler.client = CuratorFrameworkFactory.newClient(config.zk, retryPolicy) |
| // scheduler.client.start() |
| // scheduler.config = config |
| // |
| // scheduler |
| // |
| // } |
| // |
| //} |
