<commit_before>package com.eigengo.lift.common

import java.net.InetAddress
import java.util.concurrent.TimeUnit

import akka.actor._
import akka.cluster.Cluster
import akka.io.IO
import com.eigengo.lift.common.MicroserviceApp.BootedNode
import com.typesafe.config.ConfigFactory
import net.nikore.etcd.EtcdClient
import net.nikore.etcd.EtcdJsonProtocol.EtcdListResponse
import spray.can.Http
import spray.routing.{HttpServiceActor, Route}

import scala.concurrent.ExecutionContext
import scala.util.{Failure, Success}

object MicroserviceApp {

  class Api(route: Route) extends HttpServiceActor {
    override def receive: Receive = runRoute(route)
  }

  trait BootedNode {
    def api: Option[ExecutionContext ⇒ Route] = None
  }

}

abstract class MicroserviceApp(microserviceName: String)(f: ActorSystem ⇒ BootedNode) extends App {

  object EtcdKeys {
    val ClusterNodes = "akka.cluster.nodes"
  }

  import com.eigengo.lift.common.MicroserviceApp._
  private val name = "Lift"
  private val log = Logger(getClass)

  // HACK: Wait for Cassandra startup.
  Thread.sleep(10000)

  def startup(): Unit = {
    val hostname = InetAddress.getLocalHost.getHostAddress
    log.info(s"Starting up microservice $microserviceName at $hostname")

    import scala.concurrent.duration._
    val config = ConfigFactory.load()
    val etcd = new EtcdClient(config.getString("etcd.url"))
    log.info(s"Config loaded; etcd expected at $etcd")

    val retry = config.getDuration("akka.cluster.retry", TimeUnit.SECONDS).seconds

    log.info("Creating the microservice's ActorSystem")
    // Create an Akka system
    val system = ActorSystem(name, config)
    val cluster = Cluster(system)
    // Register cluster MemberUp callback
    cluster.registerOnMemberUp {
      log.info(s"*********** Node ${cluster.selfAddress} booting up")
      etcd.setKey(s"${EtcdKeys.ClusterNodes}/${clusterAddressKey()}", "MemberUp")
      // boot the microservice code
      val bootedNode = f(system)
      bootedNode.api.foreach(startupApi)
      // logme!
      log.info(s"*********** Node ${cluster.selfAddress} up")
    }
    // register shutdown callback
    system.registerOnTermination(shutdown())
    // register this (cluster) actor system with etcd
    etcd.setKey(s"${EtcdKeys.ClusterNodes}/${clusterAddressKey()}", "Joining")

    joinCluster()

    def startupApi(api: ExecutionContext ⇒ Route): Unit = {
      val route: Route = api(system.dispatcher)
      val restService = system.actorOf(Props(classOf[Api], route))
      IO(Http)(system) ! Http.Bind(restService, interface = "0.0.0.0", port = 8080) // FIXME: should we use 0.0.0.0 here?
    }

    def joinCluster(): Unit = {
      import system.dispatcher

      log.info("Joining the cluster")

      if (cluster.selfRoles.contains("initial-seed")) {
        // We are an initial seed node, so we only need to join our own cluster
        cluster.join(cluster.selfAddress)
      } else {
        // We are not an initial seed node, so we need to fetch cluster nodes for seeding
        etcd.listDir(EtcdKeys.ClusterNodes, recursive = false).onComplete {
          case Success(response: EtcdListResponse) ⇒
            log.debug(s"Using etcd response: $response")
            response.node.nodes match {
              // We are only interested in actor systems which have registered and recorded themselves as up
              case Some(systemNodes)
                if systemNodes.filterNot(_.key == s"/${EtcdKeys.ClusterNodes}/${clusterAddressKey()}").filter(_.value == Some("MemberUp")).nonEmpty => {

                // At least one actor system address has been retrieved from etcd - we now need to check their respective etcd states and locate up cluster seed nodes
                val seedNodes =
                  systemNodes
                    .filterNot(_.key == s"/${EtcdKeys.ClusterNodes}/${clusterAddressKey()}")
                    .filter(_.value == Some("MemberUp"))
                    .map(n => clusterAddress(n.key.stripPrefix(s"/${EtcdKeys.ClusterNodes}/")))

                log.info(s"Joining our cluster using the seed nodes: $seedNodes")
                cluster.joinSeedNodes(seedNodes)
              }

              case Some(_) ⇒
                log.warning(s"Not enough seed nodes found. Retrying in $retry")
                system.scheduler.scheduleOnce(retry)(joinCluster())
<commit_msg>Bug fix: removed potential race condition in setting etcd key to be MemberUp<commit_after>package com.eigengo.lift.common

import java.net.InetAddress
import java.util.concurrent.TimeUnit

import akka.actor._
import akka.cluster.Cluster
import akka.io.IO
import com.eigengo.lift.common.MicroserviceApp.BootedNode
import com.typesafe.config.ConfigFactory
import net.nikore.etcd.EtcdClient
import net.nikore.etcd.EtcdJsonProtocol.EtcdListResponse
import spray.can.Http
import spray.routing.{HttpServiceActor, Route}

import scala.concurrent.ExecutionContext
import scala.util.{Failure, Success}

object MicroserviceApp {

  class Api(route: Route) extends HttpServiceActor {
    override def receive: Receive = runRoute(route)
  }

  trait BootedNode {
    def api: Option[ExecutionContext ⇒ Route] = None
  }

}

abstract class MicroserviceApp(microserviceName: String)(f: ActorSystem ⇒ BootedNode) extends App {

  object EtcdKeys {
    val ClusterNodes = "akka.cluster.nodes"
  }

  import com.eigengo.lift.common.MicroserviceApp._
  private val name = "Lift"
  private val log = Logger(getClass)

  // HACK: Wait for Cassandra startup.
  Thread.sleep(10000)

  def startup(): Unit = {
    val hostname = InetAddress.getLocalHost.getHostAddress
    log.info(s"Starting up microservice $microserviceName at $hostname")

    import scala.concurrent.duration._
    val config = ConfigFactory.load()
    val etcd = new EtcdClient(config.getString("etcd.url"))
    log.info(s"Config loaded; etcd expected at $etcd")

    val retry = config.getDuration("akka.cluster.retry", TimeUnit.SECONDS).seconds

    log.info("Creating the microservice's ActorSystem")
    // Create an Akka system
    val system = ActorSystem(name, config)
    val cluster = Cluster(system)

    import system.dispatcher

    // register shutdown callback
    system.registerOnTermination(shutdown())

    // register this (cluster) actor system with etcd
    etcd.setKey(s"${EtcdKeys.ClusterNodes}/${clusterAddressKey()}", "Joining").onComplete {
      case Success(_) =>
        // Register cluster MemberUp callback
        cluster.registerOnMemberUp {
          log.info(s"*********** Node ${cluster.selfAddress} booting up")
          etcd.setKey(s"${EtcdKeys.ClusterNodes}/${clusterAddressKey()}", "MemberUp")
          // boot the microservice code
          val bootedNode = f(system)
          bootedNode.api.foreach(startupApi)
          // logme!
          log.info(s"*********** Node ${cluster.selfAddress} up")
        }
        joinCluster()

      case Failure(exn) =>
        log.error(s"Failed to set state to 'Joining' with etcd: $exn")
        shutdown()
    }

    def startupApi(api: ExecutionContext ⇒ Route): Unit = {
      val route: Route = api(system.dispatcher)
      val restService = system.actorOf(Props(classOf[Api], route))
      IO(Http)(system) ! Http.Bind(restService, interface = "0.0.0.0", port = 8080) // FIXME: should we use 0.0.0.0 here?
    }

    def joinCluster(): Unit = {
      log.info("Joining the cluster")

      if (cluster.selfRoles.contains("initial-seed")) {
        // We are an initial seed node, so we only need to join our own cluster
        cluster.join(cluster.selfAddress)
      } else {
        // We are not an initial seed node, so we need to fetch cluster nodes for seeding
        etcd.listDir(EtcdKeys.ClusterNodes, recursive = false).onComplete {
          case Success(response: EtcdListResponse) ⇒
            log.debug(s"Using etcd response: $response")
            response.node.nodes match {
              // We are only interested in actor systems which have registered and recorded themselves as up
              case Some(systemNodes)
                if systemNodes.filterNot(_.key == s"/${EtcdKeys.ClusterNodes}/${clusterAddressKey()}").filter(_.value == Some("MemberUp")).nonEmpty => {

                // At least one actor system address has been retrieved from etcd - we now need to check their respective etcd states and locate up cluster seed nodes
                val seedNodes =
                  systemNodes
                    .filterNot(_.key == s"/${EtcdKeys.ClusterNodes}/${clusterAddressKey()}")
                    .filter(_.value == Some("MemberUp"))
                    .map(n => clusterAddress(n.key.stripPrefix(s"/${EtcdKeys.ClusterNodes}/")))

                log.info(s"Joining our cluster using the seed nodes: $seedNodes")
                cluster.joinSeedNodes(seedNodes)
              }

              case Some(_) ⇒
                log.warning(s"Not enough seed nodes found. Retrying in $retry")
                system.scheduler.scheduleOnce(retry)(joinCluster())

{'fraction_non_alphanumeric': 0.11055555555555556, 'fraction_numerical': 0.003777777777777778, 'mean_word_length': 2.9669457910973995, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '7061640', 'n_tokens_mistral': 2650, 'n_tokens_neox': 2549, 'n_words': 674}