<commit_before>package filodb.coordinator

import scala.concurrent.duration._

import akka.actor.{ActorRef, ActorSystem}
import akka.testkit.{TestKit, TestProbe}
import com.typesafe.config.Config
import com.typesafe.scalalogging.StrictLogging
import org.scalatest.concurrent.ScalaFutures

import filodb.coordinator.client.LocalClient
import filodb.core.metadata.{DataColumn, Dataset}

class FilodbClusterNodeSpec extends RunnableSpec with ScalaFutures {

  import NodeClusterActor._

  "A FiloServer Node" must {
    FiloServerApp.main(Array.empty)

    "join the cluster" in {
      TestKit.awaitCond(FiloServerApp.cluster.isJoined, 50.seconds)
    }
    "create and setup the coordinatorActor and clusterActor" in {
      val coordinatorActor = FiloServerApp.coordinatorActor
      val clusterActor = FiloServerApp.clusterActor

      implicit val system = FiloServerApp.system
      val shardPath = ActorName.shardStatusPath(FiloServerApp.cluster.selfAddress)
      val shardActor = system.actorSelection(shardPath).resolveOne(2.seconds).futureValue
      val probe = TestProbe()
      probe.send(coordinatorActor, CoordinatorRegistered(clusterActor, shardActor))
      probe.send(coordinatorActor, MiscCommands.GetClusterActor)
      probe.expectMsgPF() {
        case Some(ref: ActorRef) => ref shouldEqual clusterActor
      }

      probe.send(clusterActor, GetRefs("worker"))
      probe.expectMsgPF(2.seconds) {
        case refs =>
          refs.asInstanceOf[Set[ActorRef]].head shouldEqual FiloServerApp.coordinatorActor
      }
    }
    "shutdown cleanly" in {
      FiloServerApp.shutdown()
      TestKit.awaitCond(FiloServerApp.cluster.isTerminated, 5.seconds)
    }
  }
}

/* scala.DelayedInit issues: extends App */
object FiloServerApp extends FilodbClusterNode with StrictLogging {

  import filodb.core.metadata.Column.ColumnType._
  import net.ceedubs.ficus.Ficus._

  override val role = ClusterRole.Server

  override lazy val system = ActorSystem(systemName, AkkaSpec.settings.allConfig)

  override lazy val cluster = FilodbCluster(system)

  lazy val clusterActor = cluster.clusterSingletonProxy(roleName, withManager = true)

  lazy val client = new LocalClient(coordinatorActor)

  def main(args: Array[String]): Unit = {
    clusterActor
    coordinatorActor
    cluster.joinSeedNodes()
    cluster.kamonInit(role)

    scala.concurrent.Await.result(metaStore.initialize(), cluster.settings.InitializationTimeout)
    cluster._isInitialized.set(true)


    client
  }

  def createDatasetFromConfig(): Unit = {
    cluster.settings.DatasetDefinitions.foreach { case (datasetName, datasetConf) =>
      createDatasetFromConfig(datasetName, datasetConf)
    }
  }

  def createDatasetFromConfig(datasetName: String, config: Config): Unit = {
    val partKeys = config.as[Seq[String]]("partition-keys")
    val rowKeys = config.as[Seq[String]]("row-keys")

    val columns =
      config.as[Seq[String]]("string-columns").map(n => DataColumn(0, n, datasetName, 0, StringColumn)) ++
        config.as[Seq[String]]("double-columns").map(n => DataColumn(0, n, datasetName, 0, DoubleColumn)) ++
        config.as[Seq[String]]("long-columns").map(n => DataColumn(0, n, datasetName, 0, LongColumn)) ++
        config.as[Seq[String]]("int-columns").map(n => DataColumn(0, n, datasetName, 0, IntColumn)) ++
        config.as[Seq[String]]("map-columns").map(n => DataColumn(0, n, datasetName, 0, MapColumn))

    val dataset = Dataset(datasetName, rowKeys, partKeys)
    logger.info(s"Creating dataset $dataset with columns $columns...")
    client.createNewDataset(dataset, columns)
  }

  Runtime.getRuntime.addShutdownHook(new Thread() {
    override def run(): Unit = cluster.shutdown()
  })<commit_msg>Fixing flaky CI test.<commit_after>package filodb.coordinator

import scala.concurrent.duration._

import akka.actor.{ActorRef, ActorSystem}
import akka.testkit.{TestKit, TestProbe}
import com.typesafe.config.Config
import com.typesafe.scalalogging.StrictLogging
import org.scalatest.concurrent.ScalaFutures

import filodb.coordinator.client.LocalClient
import filodb.core.metadata.{DataColumn, Dataset}

class FilodbClusterNodeSpec extends RunnableSpec with ScalaFutures {

  import NodeClusterActor._

  "A FiloServer Node" must {
    FiloServerApp.main(Array.empty)

    "join the cluster" in {
      TestKit.awaitCond(FiloServerApp.cluster.isJoined, 50.seconds)
    }
    "create and setup the coordinatorActor and clusterActor" in {
      val coordinatorActor = FiloServerApp.coordinatorActor
      val clusterActor = FiloServerApp.clusterActor

      implicit val system = FiloServerApp.system
      val shardPath = ActorName.shardStatusPath(FiloServerApp.cluster.selfAddress)
      val shardActor = system.actorSelection(shardPath).resolveOne(2.seconds).futureValue
      val probe = TestProbe()
      probe.send(coordinatorActor, CoordinatorRegistered(clusterActor, shardActor))
      probe.send(coordinatorActor, MiscCommands.GetClusterActor)
      probe.expectMsgPF() {
        case Some(ref: ActorRef) => ref shouldEqual clusterActor
      }

      probe.send(clusterActor, GetRefs("worker"))
      probe.expectMsgPF(2.seconds) {
        case refs =>
          refs.asInstanceOf[Set[ActorRef]].head shouldEqual FiloServerApp.coordinatorActor
      }
    }
    "shutdown cleanly" in {
      FiloServerApp.shutdown()
      TestKit.awaitCond(FiloServerApp.cluster.isTerminated, 5.seconds)
    }
  }
}

/* scala.DelayedInit issues: extends App */
object FiloServerApp extends FilodbClusterNode with StrictLogging {

  import filodb.core.metadata.Column.ColumnType._
  import net.ceedubs.ficus.Ficus._

  override val role = ClusterRole.Server

  override lazy val system = ActorSystem(systemName, AkkaSpec.settings.allConfig)

  override val cluster = FilodbCluster(system)

  val clusterActor = cluster.clusterSingletonProxy(roleName, withManager = true)
  cluster.joinSeedNodes()

  lazy val client = new LocalClient(coordinatorActor)

  def main(args: Array[String]): Unit = {
    coordinatorActor
    cluster.kamonInit(role)
    scala.concurrent.Await.result(metaStore.initialize(), cluster.settings.InitializationTimeout)
    cluster._isInitialized.set(true)

    client
  }

  def createDatasetFromConfig(): Unit = {
    cluster.settings.DatasetDefinitions.foreach { case (datasetName, datasetConf) =>
      createDatasetFromConfig(datasetName, datasetConf)
    }
  }

  def createDatasetFromConfig(datasetName: String, config: Config): Unit = {
    val partKeys = config.as[Seq[String]]("partition-keys")
    val rowKeys = config.as[Seq[String]]("row-keys")

    val columns =
      config.as[Seq[String]]("string-columns").map(n => DataColumn(0, n, datasetName, 0, StringColumn)) ++
        config.as[Seq[String]]("double-columns").map(n => DataColumn(0, n, datasetName, 0, DoubleColumn)) ++
        config.as[Seq[String]]("long-columns").map(n => DataColumn(0, n, datasetName, 0, LongColumn)) ++
        config.as[Seq[String]]("int-columns").map(n => DataColumn(0, n, datasetName, 0, IntColumn)) ++
        config.as[Seq[String]]("map-columns").map(n => DataColumn(0, n, datasetName, 0, MapColumn))

    val dataset = Dataset(datasetName, rowKeys, partKeys)
    logger.info(s"Creating dataset $dataset with columns $columns...")
    client.createNewDataset(dataset, columns)
  }

  Runtime.getRuntime.addShutdownHook(new Thread() {
    override def run(): Unit = cluster.shutdown()
  })
{'fraction_non_alphanumeric': 0.11000403931600916, 'fraction_numerical': 0.004039316009155783, 'mean_word_length': 4.283072546230441, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '7393611', 'n_tokens_mistral': 2301, 'n_tokens_neox': 2236, 'n_words': 436}