first commit

.gitignore

@@ -0,0 +1,13 @@
+target
+dist
+#*#
+.#*
+*.log
+*.iml
+*.ipr
+*.iws
+*.swp
+lib_managed
+project/boot
+.DS_Store
+src_managed

project/build.properties

@@ -0,0 +1,10 @@
+#
+#Wed Mar 03 19:39:42 UTC 2010
+project.name=zookeeper-client
+project.organization=Twitter, Inc.
+scala.version=2.7.7
+project.version=1.0
+sbt.version=0.7.1
+def.scala.version=2.7.7
+build.scala.versions=2.7.7
+project.initialize=false

project/build/ZookeeperLoadtest.scala

@@ -0,0 +1,35 @@
+import sbt._
+import Process._
+
+
+class HawkwindProject(info: ProjectInfo) extends DefaultProject(info) {
+  // Maven repositories
+  val scalaToolsTesting = "testing.scala-tools.org" at "http://scala-tools.org/repo-releases/"
+  val powerMock = "powermock-api" at "http://powermock.googlecode.com/svn/repo/"
+  val mavenDotOrg = "repo1" at "http://repo1.maven.org/maven2/"
+  val scalaToolsReleases = "scala-tools.org" at "http://scala-tools.org/repo-releases/"
+  val reucon = "reucon" at "http://maven.reucon.com/public/"
+  val lagDotNet = "lag.net" at "http://www.lag.net/repo/"
+  val oauthDotNet = "oauth.net" at "http://oauth.googlecode.com/svn/code/maven"
+  val javaDotNet = "download.java.net" at "http://download.java.net/maven/2/"
+  val jBoss = "jboss-repo" at "http://repository.jboss.org/maven2/"
+  val nest = "nest" at "http://www.lag.net/nest/"
+
+  // dependencies
+  val vscaladoc = "org.scala-tools" % "vscaladoc" % "1.1-md-3"
+  val markdownj = "markdownj" % "markdownj" % "1.0.2b4-0.3.0"
+  val slf4jApi = "org.slf4j" % "slf4j-api" % "1.5.8"
+  val slf4jLog = "org.slf4j" % "slf4j-log4j12" % "1.5.8"
+  val log4j = "apache-log4j" % "log4j" % "1.2.15"
+  val commonsLogging = "commons-logging" % "commons-logging" % "1.1"
+  val commonsLang = "commons-lang" % "commons-lang" % "2.2"
+  val oro = "oro" % "oro" % "2.0.7"
+  val configgy = "net.lag" % "configgy" % "1.4.7"
+  val mockito = "org.mockito" % "mockito-core" % "1.8.1"
+  val xrayspecs = "com.twitter" % "xrayspecs" % "1.0.5"
+  val hamcrest = "org.hamcrest" % "hamcrest-all" % "1.1"
+  val asm = "asm" % "asm-all" % "2.2"
+  val objenesis = "org.objenesis" % "objenesis" % "1.1"
+  val javautils = "org.scala-tools" % "javautils" % "2.7.4-0.1"
+  val ostrich = "com.twitter" % "ostrich" % "1.1.6"
+}

src/main/resources/config.conf

@@ -0,0 +1,17 @@
+log {
+  filename = "/tmp/zookeeperloadtest.log"
+  roll = "daily"
+  level = "debug"
+}
+
+# where test run log files are dumped
+test-runs-path = "/tmp/zookeeperloadtest"
+
+# how many get operations to do per worker per test run
+gets = 10000
+
+# how many concurrent workers to use
+concurrency = 4
+
+# where do the Zookeeper servers live?
+hostlist = "sjc1k029:2181,sjc1k030:2181,sjc1k031:2181"

src/main/scala/com/twitter/zookeeper/FakeWatcher.scala

@@ -0,0 +1,10 @@
+package com.twitter.zookeeperloadtest
+
+import org.apache.zookeeper.{Watcher, WatchedEvent}
+
+
+class FakeWatcher extends Watcher {
+  def process(event: WatchedEvent) {
+    // nop
+  }
+}

src/main/scala/com/twitter/zookeeper/LoadTest.scala

@@ -0,0 +1,45 @@
+package com.twitter.zookeeper.client.test
+
+import com.twitter.jackhammer.LoggingLoadTest
+import net.lag.configgy.{Config, Configgy}
+import net.lag.logging.Logger
+
+object LoadTest extends LoggingLoadTest {
+  Configgy.configure("src/main/resources/config.conf")
+
+  private val config = Configgy.config
+  private val log = Logger.get
+
+  val GETS = config.getInt("times-to-get", 10000)
+  val CONCURRENCY = config.getInt("concurrency", 4)
+  val HOSTLIST = config.getString("hostlist", "localhost:2181")
+
+  val watcher = new FakeWatcher
+  val zkClient = new ZookeeperClient(watcher, HOSTLIST)
+
+  def singleClientTest {
+    runWithTimingNTimes(GETS) {
+      zkClient.get("/")
+    }
+  }
+
+  def parallelClientTest {
+    runInParallelNTimes(CONCURRENCY, GETS) {
+      zkClient.get("/")
+    }
+  }
+
+  def main(args: Array[String]) {
+    log.info("starting up")
+
+    // run tests
+    parallelClientTest
+
+    // gather up results and flush them to a file
+    dumpLogOutput
+
+    // peace out
+    log.info("done with tests, exiting")
+    exit(0)
+  }
+}

src/main/scala/com/twitter/zookeeper/client/ZookeeperClient.scala

@@ -0,0 +1,72 @@
+package com.twitter.zookeeperloadtest
+
+import org.scala_tools.javautils.Imports._
+import net.lag.configgy.{Config, Configgy}
+import net.lag.logging.Logger
+import org.apache.zookeeper.{CreateMode, KeeperException, Watcher, ZooKeeper}
+import org.apache.zookeeper.data.{ACL, Stat}
+
+
+class ZookeeperClient(watcher: Watcher, hostnamePortPairs: String) {
+  Configgy.configure("src/main/resources/config.conf")
+
+  private val log = Logger.get
+  private val config = Configgy.config
+
+  val sessionTimeout = config.getInt("session-timeout", 3000)
+
+  private lazy val zkClient = new ZooKeeper(hostnamePortPairs, sessionTimeout, watcher)
+
+  def close {
+    send { zk => zk.close() }
+  }
+
+  def get(path: String): Array[Byte] = {
+    send { zk =>
+      val stat: Stat = zk.exists(path, false)
+      zk.getData(path, false, stat)
+    }
+  }
+
+  // FIXME update to 2.8 Java collection conversions
+  def getChildren(path: String): Seq[String] = {
+    send { zk =>
+      zk.getChildren(path, false).asScala
+    }
+  }
+
+  def create(path: String, data: Array[Byte], acl: java.util.List[ACL], createMode: CreateMode): String = {
+    send { zk =>
+      zk.create(path, data, acl, createMode)
+    }
+  }
+
+  def delete(path: String) {
+    send { zk =>
+      zk.delete(path, -1)
+    }
+  }
+
+  def isAlive: Boolean = {
+    val result: Stat = send { zk =>
+      zk.exists("/", false)  // do not watch
+    }
+
+    if (result.getVersion >= 0) {
+      true
+    } else {
+      false
+    }
+  }
+
+  private def send[T](f: ZooKeeper => T): T = {
+    try {
+      f(zkClient)
+    } catch {
+      case e: KeeperException => {
+        log.error(e, "Error performing Zookeeper operation")
+        throw e
+      }
+    }
+  }
+}

src/test/scala/com/twitter/zookeeper/client/ZookeeperClientSpec.scala

@@ -0,0 +1,56 @@
+package com.twitter.zookeeperloadtest
+
+import org.scala_tools.javautils.Imports._
+import org.apache.zookeeper.{CreateMode, Watcher, WatchedEvent}
+import org.apache.zookeeper.CreateMode._
+import org.apache.zookeeper.KeeperException.NoNodeException
+import org.apache.zookeeper.data.{ACL, Id}
+import org.specs._
+
+
+object ZookeeperClientSpec extends Specification {
+  "ZookeeperClient" should {
+    val watcher = new FakeWatcher
+    val zkClient = new ZookeeperClient(watcher, "localhost:2181")
+
+    doLast {
+      zkClient.close
+    }
+
+    // TODO need a doFirst to ensure that a Zookeeper server is running
+    // before proceeding with the other tests, ala Robey's Ostrich stuff
+
+    "be able to be instantiated with a FakeWatcher" in {
+      zkClient mustNot beNull
+    }
+
+    "connect to local Zookeeper server and retrieve version" in {
+      zkClient.isAlive mustBe true
+    }
+
+    "get data at a known-good specified path" in {
+      val results: Array[Byte] = zkClient.get("/")
+      results.size must beGreaterThanOrEqualTo(0)
+    }
+
+    "get data at a known-bad specified path" in {
+      zkClient.get("/thisdoesnotexist") must throwA[NoNodeException]
+    }
+
+    "get list of children" in {
+      zkClient.getChildren("/") must notBeEmpty
+    }
+
+    "create a node at a specified path" in {
+      val data: Array[Byte] = Array(0x63)
+      val id = new Id("world", "anyone")
+      val acl = new ACL(0, id)
+      val aclList = List[ACL](acl).asJava
+      val createMode = EPHEMERAL
+
+      zkClient.create("/foo", data, aclList, createMode) mustEqual "/foo"
+      zkClient.delete("/foo")
+    }
+  }
+}
+