<commit_before>package com.bwsw.sj.crud.rest.instance

import java.net.URI

import com.bwsw.common.LeaderLatch
import com.bwsw.sj.common.DAL.ConnectionConstants
import com.bwsw.sj.common.DAL.model.module.Instance
import com.bwsw.sj.common.DAL.model.{TStreamSjStream, ZKService}
import com.bwsw.sj.common.config.ConfigurationSettingsUtils
import com.bwsw.sj.common.rest.entities.MarathonRequest
import com.bwsw.sj.common.utils._
import com.bwsw.sj.crud.rest.RestLiterals
import org.apache.http.client.methods.CloseableHttpResponse
import org.slf4j.LoggerFactory

import scala.collection.JavaConversions._

/**
 * One-thread starting object for instance
 * using synchronous apache http client
 *
 *
 * @author <NAME>
 */
class InstanceStarter(instance: Instance, delay: Long = 1000) extends Runnable with InstanceManager {

  import EngineLiterals._

  private val logger = LoggerFactory.getLogger(getClass.getName)
  private lazy val restHost = ConfigurationSettingsUtils.getCrudRestHost()
  private lazy val restPort = ConfigurationSettingsUtils.getCrudRestPort()
  private lazy val restAddress = new URI(s"http://$restHost:$restPort").toString

  def run() = {
    logger.debug(s"Instance: ${instance.name}. Start instance.")
    try {
      updateInstanceStatus(instance, starting)
      startInstance()
    } catch {
      case e: Exception =>
        logger.debug(s"Instance: ${instance.name}. Failed instance.")
        logger.debug(e.getMessage)
        e.printStackTrace()
        updateInstanceStatus(instance, failed)
    }
  }

  private def startInstance() = {
    val marathonInfo = getMarathonInfo()
    if (isStatusOK(marathonInfo)) {
      val marathonMaster = getMarathonMaster(marathonInfo)
      val leaderLatch = createLeaderLatch(marathonMaster)
      startGenerators()
      tryToStartFramework(marathonMaster)
      leaderLatch.close()
    } else {
      updateInstanceStatus(instance, failed)
    }
  }

  private def createLeaderLatch(marathonMaster: String) = {
    val zkServers = getZooKeeperServers(marathonMaster)
    val leader = new LeaderLatch(Set(zkServers), RestLiterals.masterNode)
    leader.start()
    leader.takeLeadership(delay)

    leader
  }

  private def getZooKeeperServers(marathonMaster: String) = {
    var zooKeeperServers = ""
    val zkHost = System.getenv("ZOOKEEPER_HOST")
    val zkPort = System.getenv("ZOOKEEPER_PORT")
    if (zkHost != null && zkPort != null) {
      zooKeeperServers = zkHost + ":" + zkPort
    } else {
      val marathonMasterUrl = new URI(marathonMaster)
      zooKeeperServers = marathonMasterUrl.getHost + ":" + marathonMasterUrl.getPort
    }

    zooKeeperServers
  }

  private def startGenerators() = {
    val generators = getGeneratorsToStart()
    generators.foreach(startGenerator)
  }

  private def getGeneratorsToStart() = {
    val streamsHavingGenerator = getStreamsHavingGenerator(instance)
    val generatorsToStart = streamsHavingGenerator.filter(isGeneratorAvailableForStarting)

    generatorsToStart
  }

  private def isGeneratorAvailableForStarting(stream: TStreamSjStream) = {
    hasGeneratorToHandle(stream.name) || hasGeneratorFailed(instance, stream.name)
  }

  private def hasGeneratorToHandle(name: String) = {
    val stage = instance.stages.get(name)
    stage.state.equals(toHandle)
  }

  private def startGenerator(stream: TStreamSjStream) = {
    updateGeneratorState(instance, stream.name, starting)
    val applicationID = getGeneratorApplicationID(stream)
    val generatorApplicationInfo = getApplicationInfo(applicationID)
    if (isStatusOK(generatorApplicationInfo)) {
      launchGenerator(stream, applicationID)
    } else {
      createGenerator(stream, applicationID)
    }
  }

  private def launchGenerator(stream: TStreamSjStream, applicationID: String) = {
    val instanceCount = stream.generator.instanceCount
    val response = scaleMarathonApplication(applicationID, instanceCount)
    if (isStatusOK(response)) {
      waitForGeneratorToStart(applicationID, stream.name, instanceCount)
    } else {
      updateGeneratorState(instance, stream.name, failed)
    }
  }

  private def hasGeneratorStarted(response: CloseableHttpResponse, instanceCount: Int) = {
    val tasksRunning = getNumberOfRunningTasks(response)

    tasksRunning == instanceCount
  }

  private def createGenerator(stream: TStreamSjStream, applicationID: String) = {
    val request = createRequestForGeneratorCreation(stream, applicationID)
    val response = startMarathonApplication(request)
    if (isStatusCreated(response)) {
      waitForGeneratorToStart(applicationID, stream.name, stream.generator.instanceCount)
    } else {
      updateGeneratorState(instance, stream.name, failed)
    }
  }

  private def createRequestForGeneratorCreation(stream: TStreamSjStream, applicationID: String) = {
    val transactionGeneratorJar = ConfigurationSettingsUtils.getTransactionGeneratorJarName()
    val command = "java -jar " + transactionGeneratorJar + " $PORT"
    val restUrl = new URI(s"$restAddress/v1/custom/jars/$transactionGeneratorJar")
    val environmentVariables = getGeneratorEnvironmentVariables(stream)
    val marathonRequest = MarathonRequest(
      applicationID,
      command,
      stream.generator.instanceCount,
      environmentVariables,
      List(restUrl.toString))

    marathonRequest
  }

  private def getGeneratorEnvironmentVariables(stream: TStreamSjStream) = {
    val zkService = stream.generator.service.asInstanceOf[ZKService]
    val generatorProvider = zkService.provider
    val prefix = createZookeeperPrefix(zkService.namespace, stream.generator.generatorType, stream.name)

    Map("ZK_SERVERS" -> generatorProvider.hosts.mkString(";"), "PREFIX" -> prefix)
  }

  private def createZookeeperPrefix(namespace: String, generatorType: String, name: String) = {
    var prefix = namespace<commit_msg>SJ- fix transaction generator, now it uses MONGO_HOSTS<commit_after>package com.bwsw.sj.crud.rest.instance

import java.net.URI

import com.bwsw.common.LeaderLatch
import com.bwsw.sj.common.DAL.ConnectionConstants
import com.bwsw.sj.common.DAL.model.module.Instance
import com.bwsw.sj.common.DAL.model.{TStreamSjStream, ZKService}
import com.bwsw.sj.common.config.ConfigurationSettingsUtils
import com.bwsw.sj.common.rest.entities.MarathonRequest
import com.bwsw.sj.common.utils._
import com.bwsw.sj.crud.rest.RestLiterals
import org.apache.http.client.methods.CloseableHttpResponse
import org.slf4j.LoggerFactory

import scala.collection.JavaConversions._

/**
 * One-thread starting object for instance
 * using synchronous apache http client
 *
 *
 * @author <NAME>
 */
class InstanceStarter(instance: Instance, delay: Long = 1000) extends Runnable with InstanceManager {

  import EngineLiterals._

  private val logger = LoggerFactory.getLogger(getClass.getName)
  private lazy val restHost = ConfigurationSettingsUtils.getCrudRestHost()
  private lazy val restPort = ConfigurationSettingsUtils.getCrudRestPort()
  private lazy val restAddress = new URI(s"http://$restHost:$restPort").toString

  def run() = {
    logger.debug(s"Instance: ${instance.name}. Start instance.")
    try {
      updateInstanceStatus(instance, starting)
      startInstance()
    } catch {
      case e: Exception =>
        logger.debug(s"Instance: ${instance.name}. Failed instance.")
        logger.debug(e.getMessage)
        e.printStackTrace()
        updateInstanceStatus(instance, failed)
    }
  }

  private def startInstance() = {
    val marathonInfo = getMarathonInfo()
    if (isStatusOK(marathonInfo)) {
      val marathonMaster = getMarathonMaster(marathonInfo)
      val leaderLatch = createLeaderLatch(marathonMaster)
      startGenerators()
      tryToStartFramework(marathonMaster)
      leaderLatch.close()
    } else {
      updateInstanceStatus(instance, failed)
    }
  }

  private def createLeaderLatch(marathonMaster: String) = {
    val zkServers = getZooKeeperServers(marathonMaster)
    val leader = new LeaderLatch(Set(zkServers), RestLiterals.masterNode)
    leader.start()
    leader.takeLeadership(delay)

    leader
  }

  private def getZooKeeperServers(marathonMaster: String) = {
    var zooKeeperServers = ""
    val zkHost = System.getenv("ZOOKEEPER_HOST")
    val zkPort = System.getenv("ZOOKEEPER_PORT")
    if (zkHost != null && zkPort != null) {
      zooKeeperServers = zkHost + ":" + zkPort
    } else {
      val marathonMasterUrl = new URI(marathonMaster)
      zooKeeperServers = marathonMasterUrl.getHost + ":" + marathonMasterUrl.getPort
    }

    zooKeeperServers
  }

  private def startGenerators() = {
    val generators = getGeneratorsToStart()
    generators.foreach(startGenerator)
  }

  private def getGeneratorsToStart() = {
    val streamsHavingGenerator = getStreamsHavingGenerator(instance)
    val generatorsToStart = streamsHavingGenerator.filter(isGeneratorAvailableForStarting)

    generatorsToStart
  }

  private def isGeneratorAvailableForStarting(stream: TStreamSjStream) = {
    hasGeneratorToHandle(stream.name) || hasGeneratorFailed(instance, stream.name)
  }

  private def hasGeneratorToHandle(name: String) = {
    val stage = instance.stages.get(name)
    stage.state.equals(toHandle)
  }

  private def startGenerator(stream: TStreamSjStream) = {
    updateGeneratorState(instance, stream.name, starting)
    val applicationID = getGeneratorApplicationID(stream)
    val generatorApplicationInfo = getApplicationInfo(applicationID)
    if (isStatusOK(generatorApplicationInfo)) {
      launchGenerator(stream, applicationID)
    } else {
      createGenerator(stream, applicationID)
    }
  }

  private def launchGenerator(stream: TStreamSjStream, applicationID: String) = {
    val instanceCount = stream.generator.instanceCount
    val response = scaleMarathonApplication(applicationID, instanceCount)
    if (isStatusOK(response)) {
      waitForGeneratorToStart(applicationID, stream.name, instanceCount)
    } else {
      updateGeneratorState(instance, stream.name, failed)
    }
  }

  private def hasGeneratorStarted(response: CloseableHttpResponse, instanceCount: Int) = {
    val tasksRunning = getNumberOfRunningTasks(response)

    tasksRunning == instanceCount
  }

  private def createGenerator(stream: TStreamSjStream, applicationID: String) = {
    val request = createRequestForGeneratorCreation(stream, applicationID)
    val response = startMarathonApplication(request)
    if (isStatusCreated(response)) {
      waitForGeneratorToStart(applicationID, stream.name, stream.generator.instanceCount)
    } else {
      updateGeneratorState(instance, stream.name, failed)
    }
  }

  private def createRequestForGeneratorCreation(stream: TStreamSjStream, applicationID: String) = {
    val transactionGeneratorJar = ConfigurationSettingsUtils.getTransactionGeneratorJarName()
    val command = "java -jar " + transactionGeneratorJar + " $PORT"
    val restUrl = new URI(s"$restAddress/v1/custom/jars/$transactionGeneratorJar")
    val environmentVariables = getGeneratorEnvironmentVariables(stream)
    val marathonRequest = MarathonRequest(
      applicationID,
      command,
      stream.generator.instanceCount,
      environmentVariables,
      List(restUrl.toString))

    marathonRequest
  }

  private def getGeneratorEnvironmentVariables(stream: TStreamSjStream) = {
    val zkService = stream.generator.service.asInstanceOf[ZKService]
    val generatorProvider = zkService.provider
    val prefix = createZookeeperPrefix(zkService.namespace, stream.generator.generatorType, stream.name)

    Map("ZK_SERVERS" -> generatorProvider.hosts.mkString(";"),
      "PREFIX" -> prefix,
      "MONGO_HOSTS" -> ConnectionConstants.mongoHosts.map(hostport=>hostport.getHost+":"+hostport.getPort.toString).mkString(","))
  }

  private def createZookeeperPrefix(namespace: String, generatorType: String, name: String) = {
    var prefix = namespace
{'fraction_non_alphanumeric': 0.09151372284896447, 'fraction_numerical': 0.0010102710894089914, 'mean_word_length': 4.416780665754674, 'pattern_counts': {'":': 5, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5741558', 'n_tokens_mistral': 3472, 'n_tokens_neox': 3368, 'n_words': 658}