<commit_before>package com.twitter.finagle.zookeeper

import com.twitter.conversions.time._
import com.twitter.finagle.{Announcer, Resolver, Addr}
import com.twitter.util.{Await, Duration, Var}
import java.net.InetSocketAddress
import org.junit.runner.RunWith
import org.scalatest.concurrent.Eventually._
import org.scalatest.junit.JUnitRunner
import org.scalatest.time._
import org.scalatest.{BeforeAndAfter, FunSuite, Tag}
import org.scalatest.exceptions.TestFailedDueToTimeoutException
import java.net.URL
import java.io.{InputStreamReader, BufferedReader}

@RunWith(classOf[JUnitRunner])
class ZkAnnouncerTest extends FunSuite with BeforeAndAfter {
  val zkTimeout = 100.milliseconds
  var inst: ZkInstance = _
  val factory = new ZkClientFactory(zkTimeout)

  implicit val patienceConfig = PatienceConfig(
    timeout = toSpan(zkTimeout*3),
    interval = toSpan(zkTimeout))

  before {
    inst = new ZkInstance
    inst.start()
  }

  after {
    inst.stop()
  }

  def toSpan(d: Duration): Span = Span(d.inNanoseconds, Nanoseconds)
  def hostPath = "localhost:%d!/foo/bar/baz".format(inst.zookeeperAddress.getPort)

  // TODO: remove when no longer flaky.
  override def test(testName: String, testTags: Tag*)(f: => Unit) {
    if (!sys.props.contains("SKIP_FLAKY"))
      super.test(testName, testTags:_*)(f)
  }

  test("announce a primary endpoint") {
    val ann = new ZkAnnouncer(factory)
    val res = new ZkResolver(factory)
    val addr = new InetSocketAddress(8080)
    Await.result(ann.announce(addr, "%s!0".format(hostPath)))

    val va = res.bind(hostPath)
    eventually {
      Var.sample(va) match {
        case Addr.Bound(sockaddrs) =>
          assert(sockaddrs === Set(addr))
        case _ => fail()
      }
    }
  }

  test("only announce additional endpoints if a primary endpoint is present") {
    var va1: Var[Addr] = null
    var va2: Var[Addr] = null
    var failedEventually = 1

    try {
      val ann = new ZkAnnouncer(factory)
      val res = new ZkResolver(factory)
      val addr1 = new InetSocketAddress(8080)
      val addr2 = new InetSocketAddress(8081)

      Await.ready(ann.announce(addr2, "%s!0!addr2".format(hostPath)))
      va2 = res.bind("%s!addr2".format(hostPath))
      eventually { assert(Var.sample(va2) != Addr.Pending) }
      failedEventually += 1
      assert(Var.sample(va2) === Addr.Neg)

      Await.ready(ann.announce(addr1, "%s!0".format(hostPath)))
      va1 = res.bind(hostPath)
      eventually { assert(Var.sample(va2) === Addr.Bound(addr2)) }
      failedEventually += 1
      eventually { assert(Var.sample(va1) === Addr.Bound(addr1)) }
    } catch {
      case e: TestFailedDueToTimeoutException =>
        var exceptionString = "#%d eventually failed.\n".format(failedEventually)

        if(va1 != null) {
          exceptionString += "va1 status: %s\n".format(Var.sample(va1).toString)
        }

        if(va2 != null) {
          exceptionString += "va2 status: %s\n".format(Var.sample(va2).toString)
        }

        val endpoint = "/services/ci"
        val connection = new URL("http", "0.0.0.0", 4860, endpoint).openConnection()
        val reader = new BufferedReader(new InputStreamReader(connection.getInputStream))
        var fullOutput = ""
        var line = reader.readLine()
        while (line != null) {
          fullOutput += line
          line = reader.readLine()
        }
        exceptionString += "output from %s -- %s".format(endpoint, fullOutput)

        throw new Exception(exceptionString)
    }
  }

  test("unannounce additional endpoints, but not primary endpoints") {
    val ann = new ZkAnnouncer(factory)
    val res = new ZkResolver(factory)
    val addr1 = new InetSocketAddress(8080)
    val addr2 = new InetSocketAddress(8081)<commit_msg>[split]  Adding monitoring/debugging for ZKAnnouncerTest "only announce additional endpoints if a primary endpoint is present"  This test is currently marked as flaky and doesn't run in any CI run, except birdcage_flakies (CI run that runs flaky tests).  It regularly fails in birdcage_flakies runs, which makes it a prime test case for testing monitoring.  The results from the output of this will be used to fine tune what stats we're printing or making available in terms of port monitoring on CI.<commit_after>package com.twitter.finagle.zookeeper

import com.twitter.conversions.time._
import com.twitter.finagle.{Announcer, Resolver, Addr}
import com.twitter.util.{Await, Duration, Var}
import java.net.InetSocketAddress
import org.junit.runner.RunWith
import org.scalatest.concurrent.Eventually._
import org.scalatest.junit.JUnitRunner
import org.scalatest.time._
import org.scalatest.{BeforeAndAfter, FunSuite, Tag}
import org.scalatest.exceptions.TestFailedDueToTimeoutException
import java.net.URL
import java.io.{InputStreamReader, BufferedReader}

@RunWith(classOf[JUnitRunner])
class ZkAnnouncerTest extends FunSuite with BeforeAndAfter {
  val zkTimeout = 100.milliseconds
  var inst: ZkInstance = _
  val factory = new ZkClientFactory(zkTimeout)

  implicit val patienceConfig = PatienceConfig(
    timeout = toSpan(zkTimeout*3),
    interval = toSpan(zkTimeout))

  before {
    inst = new ZkInstance
    inst.start()
  }

  after {
    inst.stop()
  }

  def toSpan(d: Duration): Span = Span(d.inNanoseconds, Nanoseconds)
  def hostPath = "localhost:%d!/foo/bar/baz".format(inst.zookeeperAddress.getPort)

  // TODO: remove when no longer flaky.
  override def test(testName: String, testTags: Tag*)(f: => Unit) {
    if (!sys.props.contains("SKIP_FLAKY"))
      super.test(testName, testTags:_*)(f)
  }

  test("announce a primary endpoint") {
    val ann = new ZkAnnouncer(factory)
    val res = new ZkResolver(factory)
    val addr = new InetSocketAddress(8080)
    Await.result(ann.announce(addr, "%s!0".format(hostPath)))

    val va = res.bind(hostPath)
    eventually {
      Var.sample(va) match {
        case Addr.Bound(sockaddrs) =>
          assert(sockaddrs === Set(addr))
        case _ => fail()
      }
    }
  }

  test("only announce additional endpoints if a primary endpoint is present") {
    var va1: Var[Addr] = null
    var va2: Var[Addr] = null
    var failedEventually = 1

    try {
      val ann = new ZkAnnouncer(factory)
      val res = new ZkResolver(factory)
      val addr1 = new InetSocketAddress(8080)
      val addr2 = new InetSocketAddress(8081)

      Await.ready(ann.announce(addr2, "%s!0!addr2".format(hostPath)))
      va2 = res.bind("%s!addr2".format(hostPath))
      eventually { assert(Var.sample(va2) != Addr.Pending) }
      failedEventually += 1
      assert(Var.sample(va2) === Addr.Neg)

      Await.ready(ann.announce(addr1, "%s!0".format(hostPath)))
      va1 = res.bind(hostPath)
      eventually { assert(Var.sample(va2) === Addr.Bound(addr2)) }
      failedEventually += 1
      eventually { assert(Var.sample(va1) === Addr.Bound(addr1)) }
    } catch {
      case e: TestFailedDueToTimeoutException =>
        var exceptionString = "#%d eventually failed.\n".format(failedEventually)

        if(va1 != null) {
          exceptionString += "va1 status: %s\n".format(Var.sample(va1).toString)
        }

        if(va2 != null) {
          exceptionString += "va2 status: %s\n".format(Var.sample(va2).toString)
        }

        val endpoint = "/services/ci"
        val connection = new URL("http", "0.0.0.0", 4680, endpoint).openConnection()
        val reader = new BufferedReader(new InputStreamReader(connection.getInputStream))
        var fullOutput = ""
        var line = reader.readLine()
        while (line != null) {
          fullOutput += line
          line = reader.readLine()
        }
        exceptionString += "output from %s -- %s".format(endpoint, fullOutput)

        throw new Exception(exceptionString)
    }
  }

  test("unannounce additional endpoints, but not primary endpoints") {
    val ann = new ZkAnnouncer(factory)
    val res = new ZkResolver(factory)
    val addr1 = new InetSocketAddress(8080)
    val addr2 = new InetSocketAddress(8081)
{'fraction_non_alphanumeric': 0.11614837017609592, 'fraction_numerical': 0.015486449356812789, 'mean_word_length': 3.3333333333333335, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '2773889', 'n_tokens_mistral': 2599, 'n_tokens_neox': 2494, 'n_words': 602}