finagle-integration => finagle-stress

finagle-core/src/main/scala/com/twitter/finagle/stats/StatsReceiver.scala

@@ -3,4 +3,4 @@ package com.twitter.finagle.stats
 trait StatsReceiver {
   def observer(prefix: String, label: String): (Seq[String], Int, Int) => Unit
   def makeGauge(name: String, f: => Float)
-}
+}

finagle-integration/src/main/scala/com/twitter/finagle/integration/EmbeddedServer.scala → finagle-stress/src/main/scala/com/twitter/finagle/integration/EmbeddedServer.scala

@@ -28,7 +28,7 @@ object EmbeddedServer {
 class EmbeddedServer(val addr: SocketAddress) {
   import EmbeddedServer._
 
-  // (Publically accessible) stats covering this server.
+  // (Publicly accessible) stats covering this server.
   val stats = new StatsCollection
 
   // Server state:
@@ -86,6 +86,10 @@ class EmbeddedServer(val addr: SocketAddress) {
           super.messageReceived(ctx, e)
         }
 
+        override def exceptionCaught(ctx: ChannelHandlerContext, e: ExceptionEvent) {
+          stats.incr("exc_%s".format(e.getCause.getClass.getName.split('.').last))
+        }
+
       })
 
       pipeline.addLast("latency", new SimpleChannelDownstreamHandler {

finagle-stress/src/main/scala/com/twitter/finagle/integration/LoadBalancerTest.scala

@@ -0,0 +1,107 @@
+package com.twitter.finagle.integration
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.jboss.netty.handler.codec.http._
+
+import com.twitter.ostrich.{StatsCollection, StatsProvider}
+import com.twitter.util.{Duration, CountDownLatch, Return, Throw, Time}
+import com.twitter.conversions.time._
+
+import com.twitter.finagle.builder.{ClientBuilder, Http}
+import com.twitter.finagle.service.Service
+
+object LoadBalancerTest {
+  def main(args: Array[String]) {
+    runSuite(
+      ClientBuilder()
+        .requestTimeout(10.milliseconds)
+    )
+
+    // TODO: proper resource releasing, etc.
+  }
+
+  def runSuite(clientBuilder: ClientBuilder) {
+    val baseline = new LoadBalancerTest(clientBuilder)({ case _ => })
+    println("** BASELINE")
+    baseline.run()
+  }
+
+}
+
+class LoadBalancerTest(
+  clientBuilder: ClientBuilder,
+  serverLatency: Duration = 0.seconds,
+  numRequests: Int = 100000,
+  concurrency: Int = 50)(behavior: PartialFunction[(Int, Seq[EmbeddedServer]), Unit])
+{
+  private[this] val requestNumber = new AtomicInteger(0)
+  private[this] val requestCount  = new AtomicInteger(numRequests)
+  private[this] val latch         = new CountDownLatch(concurrency)
+  private[this] val stats         = new StatsCollection
+
+  private[this] def prettyPrintStats(stats: StatsProvider) {
+    stats.getCounterStats foreach { case (name, count) =>
+      println("# %-30s %d".format(name, count))
+    }
+  }
+
+  private[this] def dispatch(
+      client: Service[HttpRequest, HttpResponse],
+      servers: Seq[EmbeddedServer],
+      f: PartialFunction[(Int, Seq[EmbeddedServer]), Unit]) {
+    val num = requestNumber.incrementAndGet()
+    if (f.isDefinedAt((num, servers)))
+      f((num, servers))
+
+    client(new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/")) respond { result =>
+      result match {
+        case Return(_) =>
+          stats.incr("success")
+        case Throw(exc) =>
+          stats.incr("fail")
+          stats.incr("fail_%s".format(exc.getClass.getName.split('.').last))
+      }
+
+      if (requestCount.decrementAndGet() > 0)
+        dispatch(client, servers, f)
+      else
+        latch.countDown()
+    }
+  }
+
+  def run() {
+    val servers = (0 until 3).toArray map(_ => EmbeddedServer())
+
+    servers foreach { server =>
+      server.setLatency(serverLatency)
+    }
+
+    val client = clientBuilder
+      .codec(Http)
+      .hosts(servers map(_.addr))
+      .buildService[HttpRequest, HttpResponse]
+
+    val begin = Time.now
+    0 until concurrency foreach { _ => dispatch(client, servers, behavior) }
+    latch.await()
+    val duration = begin.untilNow
+    val rps = (numRequests.toDouble / duration.inMilliseconds.toDouble) * 1000.0
+
+    // Produce a "report" here instead, so we have some sort of
+    // semantic information here.
+
+    println("> STATS")
+    val succ = stats.getCounter("success")().toDouble
+    val fail = stats.getCounter("fail")().toDouble
+    println("> success rate: %.2f".format(100.0 * succ / (succ + fail)))
+    println("> request rate: %.2f".format(rps))
+    prettyPrintStats(stats)
+
+    servers.zipWithIndex foreach { case (server, which) =>
+      server.stop()
+      println("> SERVER[%d]".format(which))
+      prettyPrintStats(server.stats)
+    }
+  }
+}

finagle-integration/src/test/scala/com/twitter/finagle/integration/LoadBalancerIntegrationSpec.scala → finagle-stress/src/test/scala/com/twitter/finagle/integration/LoadBalancerIntegrationSpec.scala

@@ -7,7 +7,7 @@ import org.jboss.netty.handler.codec.http._
 import org.specs.Specification
 
 import com.twitter.conversions.time._
-import com.twitter.util.{Return, Throw, CountDownLatch}
+import com.twitter.util.{Return, Throw, CountDownLatch, Time}
 
 import com.twitter.ostrich.{StatsCollection, StatsProvider}
 
@@ -17,24 +17,25 @@ import com.twitter.finagle.builder.{ClientBuilder, Http}
 object LoadBalancerIntegrationSpec extends Specification {
   def prettyPrintStats(stats: StatsProvider) {
     stats.getCounterStats foreach { case (name, count) =>
-      println("# %-15s %d".format(name, count))
+      println("# %-30s %d".format(name, count))
     }
   }
 
   "Load Balancer" should {
+    // def runSuite(client: Service[HttpRequest, HttpResponse])
+    val numRequests = 50000
+    val concurrency = 50
+
     val servers = (0 until 3).toArray map(_ => EmbeddedServer())
     val stats = new StatsCollection
     val requestNumber = new AtomicInteger(0)
-    val requestCount = new AtomicInteger(10000)
-    val concurrency = 50
+    val requestCount = new AtomicInteger(numRequests)
     val latch = new CountDownLatch(concurrency)
 
     servers foreach { server =>
       server.setLatency(5.milliseconds)
     }
 
-    // TODO: parallelize these; measure throughput.
-
     // XXX - periodically print load, etc [or any kind of debugging
     // information from the loadbalancer]
 
@@ -57,7 +58,7 @@ object LoadBalancerIntegrationSpec extends Specification {
             stats.incr("success")
           case Throw(exc) =>
             stats.incr("fail")
-            stats.incr("fail_%s".format(exc.getClass.getName))
+            stats.incr("fail_%s".format(exc.getClass.getName.split('.').last))
         }
 
         if (requestCount.decrementAndGet() > 0)
@@ -66,21 +67,19 @@ object LoadBalancerIntegrationSpec extends Specification {
           latch.countDown()
       }
     }
-    
+
     def runTest[A](client: Service[HttpRequest, HttpResponse])(f: PartialFunction[Int, Unit]) {
+      val begin = Time.now
       0 until concurrency foreach { _ => dispatch(client, f) }
       latch.await()
-
+      val duration = begin.untilNow()
+      val rps = (numRequests.toDouble / duration.inMilliseconds.toDouble) * 1000.0
+
       println("> STATS")
       val succ = stats.getCounter("success")().toDouble
       val fail = stats.getCounter("fail")().toDouble
       println("> success rate: %.2f".format(100.0 * succ / (succ + fail)))
-
-      // val counterKeys = (Set() ++ stats.getCounterKeys) -- Set("success", "fail")
-      // counterKeys foreach { key =>
-      //   println("> %s = %d".format(key, stats.getCounter(key)()))
-      // }
-
+      println("> request rate: %.2f".format(rps))
       prettyPrintStats(stats)
     }
 
@@ -89,7 +88,7 @@ object LoadBalancerIntegrationSpec extends Specification {
         .codec(Http)
         .hosts(servers map(_.addr))
         .retries(2)
-        .requestTimeout(20.milliseconds)
+        .requestTimeout(50.milliseconds)
         .buildService[HttpRequest, HttpResponse]
 
       runTest(client) { case _ => () }
@@ -102,7 +101,7 @@ object LoadBalancerIntegrationSpec extends Specification {
         .codec(Http)
         .hosts(servers map(_.addr))
         .retries(2)
-        .requestTimeout(10.milliseconds)
+        .requestTimeout(50.milliseconds)
         .buildService[HttpRequest, HttpResponse]
 
       runTest(client) {
@@ -117,7 +116,7 @@ object LoadBalancerIntegrationSpec extends Specification {
       val client = ClientBuilder()
         .codec(Http)
         .hosts(servers map(_.addr))
-        .requestTimeout(10.milliseconds)
+        .requestTimeout(50.milliseconds)
         // .retries(2)
         .buildService[HttpRequest, HttpResponse]
 
@@ -134,7 +133,7 @@ object LoadBalancerIntegrationSpec extends Specification {
         .codec(Http)
         .hosts(servers map(_.addr))
         // .retries(2)
-        .requestTimeout(10.milliseconds)
+        .requestTimeout(50.milliseconds)
         .buildService[HttpRequest, HttpResponse]
 
       runTest(client) {
@@ -150,7 +149,7 @@ object LoadBalancerIntegrationSpec extends Specification {
         .codec(Http)
         .hosts(servers map(_.addr))
         // .retries(2)
-        .requestTimeout(10.milliseconds)
+        .requestTimeout(50.milliseconds)
         .buildService[HttpRequest, HttpResponse]
 
       runTest(client) {

project/build/Project.scala

@@ -27,9 +27,9 @@ class Project(info: ProjectInfo) extends StandardParentProject(info)
 
   // finagle-integration has integration test suites & tools for
   // development.
-  val integrationProject = project(
-    "finagle-integration", "finagle-integration",
-    new IntegrationProject(_), coreProject)
+  val stressProject = project(
+    "finagle-stress", "finagle-stress",
+    new StressProject(_), coreProject)
 
   class CoreProject(info: ProjectInfo) extends StandardProject(info)
     with SubversionPublisher with AdhocInlines
@@ -60,7 +60,7 @@ class Project(info: ProjectInfo) extends StandardParentProject(info)
     val ostrich = "com.twitter" % "ostrich" % "2.3.4"
   }
 
-  class IntegrationProject(info: ProjectInfo) extends StandardProject(info)
+  class StressProject(info: ProjectInfo) extends StandardProject(info)
     with SubversionPublisher with IntegrationSpecs with AdhocInlines
   {
     val ostrich = "com.twitter" % "ostrich" % "2.3.4"