<commit_before>/*
 * 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.gearpump.services

import java.io.{File, IOException}

import akka.actor.{ActorRef, ActorSystem}
import com.typesafe.config.Config
import org.apache.gearpump.util.FileUtils
import org.apache.gearpump.cluster.AppMasterToMaster.{GetAllWorkers, GetMasterData, GetWorkerData, MasterData, WorkerData}
import org.apache.gearpump.cluster.ClientToMaster.{QueryHistoryMetrics, QueryMasterConfig}
import org.apache.gearpump.cluster.MasterToAppMaster.{AppMastersData, AppMastersDataRequest, WorkerList}
import org.apache.gearpump.cluster.MasterToClient.{HistoryMetrics, MasterConfig, SubmitApplicationResultValue}
import org.apache.gearpump.cluster.UserConfig
import org.apache.gearpump.cluster.client.ClientContext
import org.apache.gearpump.cluster.main.AppSubmitter
import org.apache.gearpump.cluster.worker.{WorkerSummary}
import org.apache.gearpump.partitioner.{PartitionerByClassName, PartitionerDescription}
import org.apache.gearpump.streaming.StreamApplication
import org.apache.gearpump.streaming.appmaster.SubmitApplicationRequest
import org.apache.gearpump.util.ActorUtil._
import org.apache.gearpump.util.Constants.GEARPUMP_APP_JAR
import org.apache.gearpump.util.{Constants, Util}
import spray.http.{BodyPart, MediaTypes, MultipartFormData}
import spray.routing
import spray.routing.HttpService

import scala.collection.JavaConversions._
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Try, Failure, Success}
import org.apache.gearpump.services.util.UpickleUtil._

trait MasterService extends HttpService {
  import upickle.default.{read, write}
  def master:ActorRef
  implicit val system: ActorSystem

  implicit val ec: ExecutionContext = actorRefFactory.dispatcher
  implicit val timeout = Constants.FUTURE_TIMEOUT

  def masterRoute: routing.Route = {
    pathPrefix("api" / s"$REST_VERSION" / "master") {
      pathEnd {
        get {
          onComplete(askActor[MasterData](master, GetMasterData)) {
            case Success(value: MasterData) => complete(write(value))
            case Failure(ex) => failWith(ex)
          }
        }
      } ~
      path("applist") {
        onComplete(askActor[AppMastersData](master, AppMastersDataRequest)) {
          case Success(value: AppMastersData) =>
            complete(write(value))
          case Failure(ex) => failWith(ex)
        }
      } ~
      path("workerlist") {
        def future = askActor[WorkerList](master, GetAllWorkers).flatMap { workerList =>
          val workers = workerList.workers
          val workerDataList = List.empty[WorkerSummary]

          Future.fold(workers.map { workerId =>
            askWorker[WorkerData](master, workerId, GetWorkerData(workerId))
          })(workerDataList) { (workerDataList, workerData) =>
            workerDataList :+ workerData.workerDescription
          }
        }
        onComplete(future) {
          case Success(result: List[WorkerSummary]) => complete(write(result))
          case Failure(ex) => failWith(ex)
        }
      } ~
      path("config") {
        onComplete(askActor[MasterConfig](master, QueryMasterConfig)) {
          case Success(value: MasterConfig) =>
            val config = Option(value.config).map(_.root.render()).getOrElse("{}")
            complete(config)
          case Failure(ex) =>
            failWith(ex)
        }
      } ~
      path("metrics" / RestPath ) { path =>
        parameter("readLatest" ? "false") { readLatestInput =>
          val readLatest = Try(readLatestInput.toBoolean).getOrElse(false)
          val query = QueryHistoryMetrics(path.head.toString, readLatest)
          onComplete(askActor[HistoryMetrics](master, query)) {
            case Success(value) =>
              complete(write(value))
            case Failure(ex) =>
              failWith(ex)
          }
        }
      } ~
      path("submitapp") {
        post {
          anyParams('args.as[Option[String]]) { (args) =>
            respondWithMediaType(MediaTypes.`application/json`) {
              entity(as[MultipartFormData]) { formData =>
                import formData.fields
                val jar = MasterService.findFormDataOption(fields, "jar")
                if (jar.isEmpty) {
                  complete(write(
                    MasterService.Status(success=false, reason="Jar file not supplied")))
                }
                val userConf = MasterService.findFormDataOption(fields, "conf")
                val argsArray = args.getOrElse("").split(" +")

                onComplete(Future(
                  MasterService.submitJar(jar.get, userConf, argsArray, system.settings.config))) {
                  case Success(_) =>
                    complete(write(
                      MasterService.Status(success=true)))
                  case Failure(ex) =>
                    failWith(ex)
                }
              }
            }
          }
        }
      } ~
      path("submitdag") {
        post {
          entity(as[String]) { request =>
            val submitApplicationRequest = read[SubmitApplicationRequest](request)
            import submitApplicationRequest.{appJar, appName, dag, processors}

            Option(appJar) match {
              case Some(jar) =>
                System.setProperty(GEARPUMP_APP_JAR, appJar)
              case None =>
            }

            val context = ClientContext(system.settings.config, Some(system), Some(master))

            val graph = dag.mapVertex {processorId =>
              processors(processorId)
            }.mapEdge { (node1, edge, node2) =>
              PartitionerDescription(new PartitionerByClassName(edge))
            }

            val appId = context.submit(new StreamApplication(appName, UserConfig.empty, graph))

            import upickle.default.{read, write}
            val submitApplicationResultValue = SubmitApplicationResultValue(appId)
            val jsonData = write(submitApplicationResultValue)
            complete(jsonData)
          }
        }
      }
    }
  }
}

object MasterService {
  case class Status(success: Boolean, reason: String = null)

  /**
   * Upload user application (JAR) into temporary directory and use AppSubmitter to submit
   * it to master. The temporary file will be removed after submission is done/failed.
   */
  def submitJar(jarData: Array[Byte], userConfData: Option[Array[Byte]], extraArgs: Array[String],
                sysConfig: Config): Unit = {
    val jar = File.createTempFile("gearpump_userapp_", "")
    val userConf = File.createTempFile("gearpump_userconf_", "")

    try {
      val masters = sysConfig.getStringList(Constants.GEARPUMP_CLUSTER_MASTERS).toList.flatMap(Util.parseHostList)
      val master = masters.head
      val hostname = sysConfig.getString(Constants.GEARPUMP_HOSTNAME)
      var options = Array(
        s"-D${Constants.GEARPUMP_CLUSTER_MASTERS}.0=${master.host}:${master.port}",
        s"-D${Constants.GEARPUMP_HOSTNAME}=$hostname"
      )
      if (userConfData.isDefined) {
        FileUtils.writeByteArrayToFile(userConf, userConfData.get)
        options :+= s"-D${Constants.GEARPUMP_CUSTOM_CONFIG_FILE}=${userConf.getPath}"
      }

      FileUtils.writeByteArrayToFile(jar, jarData)
      val arguments = Array("-jar", jar.getPath) ++ extraArgs
      val mainClass = AppSubmitter.getClass.getName.dropRight(1)
      val process = Util.startProcess(options, Util.getCurrentClassPath, mainClass, arguments)
      val retval = process.exitValue()
      if (retval != 0) {
        throw new IOException(s"Process exit abnormally with code $retval, error summary: ${process.logger.summary}")
      }
    } finally {
      userConf.delete()
      jar.delete()
    }<commit_msg>fix #1341, Cannot submit application in UI if one master is killed and another master started<commit_after>/*
 * 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.gearpump.services

import java.io.{File, IOException}

import akka.actor.{ActorRef, ActorSystem}
import com.typesafe.config.Config
import org.apache.gearpump.util.FileUtils
import org.apache.gearpump.cluster.AppMasterToMaster.{GetAllWorkers, GetMasterData, GetWorkerData, MasterData, WorkerData}
import org.apache.gearpump.cluster.ClientToMaster.{QueryHistoryMetrics, QueryMasterConfig}
import org.apache.gearpump.cluster.MasterToAppMaster.{AppMastersData, AppMastersDataRequest, WorkerList}
import org.apache.gearpump.cluster.MasterToClient.{HistoryMetrics, MasterConfig, SubmitApplicationResultValue}
import org.apache.gearpump.cluster.UserConfig
import org.apache.gearpump.cluster.client.ClientContext
import org.apache.gearpump.cluster.main.AppSubmitter
import org.apache.gearpump.cluster.worker.{WorkerSummary}
import org.apache.gearpump.partitioner.{PartitionerByClassName, PartitionerDescription}
import org.apache.gearpump.streaming.StreamApplication
import org.apache.gearpump.streaming.appmaster.SubmitApplicationRequest
import org.apache.gearpump.util.ActorUtil._
import org.apache.gearpump.util.Constants.GEARPUMP_APP_JAR
import org.apache.gearpump.util.{Constants, Util}
import spray.http.{BodyPart, MediaTypes, MultipartFormData}
import spray.routing
import spray.routing.HttpService

import scala.collection.JavaConversions._
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Try, Failure, Success}
import org.apache.gearpump.services.util.UpickleUtil._

trait MasterService extends HttpService {
  import upickle.default.{read, write}
  def master:ActorRef
  implicit val system: ActorSystem

  implicit val ec: ExecutionContext = actorRefFactory.dispatcher
  implicit val timeout = Constants.FUTURE_TIMEOUT

  def masterRoute: routing.Route = {
    pathPrefix("api" / s"$REST_VERSION" / "master") {
      pathEnd {
        get {
          onComplete(askActor[MasterData](master, GetMasterData)) {
            case Success(value: MasterData) => complete(write(value))
            case Failure(ex) => failWith(ex)
          }
        }
      } ~
      path("applist") {
        onComplete(askActor[AppMastersData](master, AppMastersDataRequest)) {
          case Success(value: AppMastersData) =>
            complete(write(value))
          case Failure(ex) => failWith(ex)
        }
      } ~
      path("workerlist") {
        def future = askActor[WorkerList](master, GetAllWorkers).flatMap { workerList =>
          val workers = workerList.workers
          val workerDataList = List.empty[WorkerSummary]

          Future.fold(workers.map { workerId =>
            askWorker[WorkerData](master, workerId, GetWorkerData(workerId))
          })(workerDataList) { (workerDataList, workerData) =>
            workerDataList :+ workerData.workerDescription
          }
        }
        onComplete(future) {
          case Success(result: List[WorkerSummary]) => complete(write(result))
          case Failure(ex) => failWith(ex)
        }
      } ~
      path("config") {
        onComplete(askActor[MasterConfig](master, QueryMasterConfig)) {
          case Success(value: MasterConfig) =>
            val config = Option(value.config).map(_.root.render()).getOrElse("{}")
            complete(config)
          case Failure(ex) =>
            failWith(ex)
        }
      } ~
      path("metrics" / RestPath ) { path =>
        parameter("readLatest" ? "false") { readLatestInput =>
          val readLatest = Try(readLatestInput.toBoolean).getOrElse(false)
          val query = QueryHistoryMetrics(path.head.toString, readLatest)
          onComplete(askActor[HistoryMetrics](master, query)) {
            case Success(value) =>
              complete(write(value))
            case Failure(ex) =>
              failWith(ex)
          }
        }
      } ~
      path("submitapp") {
        post {
          anyParams('args.as[Option[String]]) { (args) =>
            respondWithMediaType(MediaTypes.`application/json`) {
              entity(as[MultipartFormData]) { formData =>
                import formData.fields
                val jar = MasterService.findFormDataOption(fields, "jar")
                if (jar.isEmpty) {
                  complete(write(
                    MasterService.Status(success=false, reason="Jar file not supplied")))
                }
                val userConf = MasterService.findFormDataOption(fields, "conf")
                val argsArray = args.getOrElse("").split(" +")

                onComplete(Future(
                  MasterService.submitJar(jar.get, userConf, argsArray, system.settings.config))) {
                  case Success(_) =>
                    complete(write(
                      MasterService.Status(success=true)))
                  case Failure(ex) =>
                    failWith(ex)
                }
              }
            }
          }
        }
      } ~
      path("submitdag") {
        post {
          entity(as[String]) { request =>
            val submitApplicationRequest = read[SubmitApplicationRequest](request)
            import submitApplicationRequest.{appJar, appName, dag, processors}

            Option(appJar) match {
              case Some(jar) =>
                System.setProperty(GEARPUMP_APP_JAR, appJar)
              case None =>
            }

            val context = ClientContext(system.settings.config, Some(system), Some(master))

            val graph = dag.mapVertex {processorId =>
              processors(processorId)
            }.mapEdge { (node1, edge, node2) =>
              PartitionerDescription(new PartitionerByClassName(edge))
            }

            val appId = context.submit(new StreamApplication(appName, UserConfig.empty, graph))

            import upickle.default.{read, write}
            val submitApplicationResultValue = SubmitApplicationResultValue(appId)
            val jsonData = write(submitApplicationResultValue)
            complete(jsonData)
          }
        }
      }
    }
  }
}

object MasterService {
  case class Status(success: Boolean, reason: String = null)

  /**
   * Upload user application (JAR) into temporary directory and use AppSubmitter to submit
   * it to master. The temporary file will be removed after submission is done/failed.
   */
  def submitJar(jarData: Array[Byte], userConfData: Option[Array[Byte]], extraArgs: Array[String],
                sysConfig: Config): Unit = {
    val jar = File.createTempFile("gearpump_userapp_", "")
    val userConf = File.createTempFile("gearpump_userconf_", "")

    try {
      val masters = sysConfig.getStringList(Constants.GEARPUMP_CLUSTER_MASTERS).toList.flatMap(Util.parseHostList)
      val mastersOption = masters.zipWithIndex.map{kv =>
        val (master, index) = kv
        s"-D${Constants.GEARPUMP_CLUSTER_MASTERS}.${index}=${master.host}:${master.port}"
      }.toArray

      val hostname = sysConfig.getString(Constants.GEARPUMP_HOSTNAME)
      var options = Array(
        s"-D${Constants.GEARPUMP_HOSTNAME}=$hostname"
      ) ++ mastersOption

      if (userConfData.isDefined) {
        FileUtils.writeByteArrayToFile(userConf, userConfData.get)
        options :+= s"-D${Constants.GEARPUMP_CUSTOM_CONFIG_FILE}=${userConf.getPath}"
      }

      FileUtils.writeByteArrayToFile(jar, jarData)
      val arguments = Array("-jar", jar.getPath) ++ extraArgs
      val mainClass = AppSubmitter.getClass.getName.dropRight(1)
      val process = Util.startProcess(options, Util.getCurrentClassPath, mainClass, arguments)
      val retval = process.exitValue()
      if (retval != 0) {
        throw new IOException(s"Process exit abnormally with code $retval, error summary: ${process.logger.summary}")
      }
    } finally {
      userConf.delete()
      jar.delete()
    }
{'fraction_non_alphanumeric': 0.10107714701601164, 'fraction_numerical': 0.001222707423580786, 'mean_word_length': 2.7941241440247406, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 52, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7153209', 'n_tokens_mistral': 4745, 'n_tokens_neox': 4653, 'n_words': 1012}