Fix bugs

.gitignore

@@ -3,3 +3,9 @@ lib_managed/
 src_managed/
 project/boot/
 project/plugins/project
+out/
+build/
+.gradle
+*.iml
+*.ipr
+*.iws

README.md

@@ -269,3 +269,4 @@ If you are building a partitioned cluster then you will want to use the `Partiti
 * maxConnectionsPerNode - the maximum number of open connections to a node. The total number of connections that can be opened by a network client is maxConnectionsPerNode * number of nodes
 * staleRequestTimeoutMins - the number of minutes to keep a request that is waiting for a response
 * staleRequestCleanupFrequenceMins - the frequency to clean up stale requests
+

build.gradle

@@ -0,0 +1,27 @@
+project.ext.isDefaultEnvironment = !project.hasProperty('overrideBuildEnvironment')
+
+File getEnvironmentScript()
+{
+  final File env = file(isDefaultEnvironment ? 'defaultEnvironment.gradle' : project.overrideBuildEnvironment)
+  assert env.isFile() : "The environment script [$env] does not exists or is not a file."
+  return env
+}
+
+apply from: environmentScript
+
+project.ext.externalDependency = [
+  'zookeeper':'org.apache.zookeeper:zookeeper:3.3.0',
+  'protobuf':'com.google.protobuf:protobuf-java:2.4.0a',
+  'log4j':'log4j:log4j:1.2.16',
+  'netty':'io.netty:netty:3.5.11.Final',
+  'slf4jApi':'org.slf4j:slf4j-api:1.5.6',
+  'slf4jLog4j':'org.slf4j:slf4j-log4j12:1.5.6',
+  'specs':'org.scala-tools.testing:specs_2.8.1:1.6.8',
+  'mockitoAll':'org.mockito:mockito-all:1.8.4',
+  'cglib':'cglib:cglib:2.1_3',
+  'objenesis':'org.objenesis:objenesis:1.2',
+  'scalaCompiler': 'org.scala-lang:scala-compiler:2.8.1',
+  'scalaLibrary': 'org.scala-lang:scala-library:2.8.1',
+  'scalatest': 'org.scalatest:scalatest:1.2'
+];
+

cluster/build.gradle

@@ -0,0 +1,18 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile externalDependency.scalaLibrary 
+  compile externalDependency.zookeeper
+  compile externalDependency.protobuf
+  compile externalDependency.log4j
+
+  testCompile externalDependency.specs
+  testCompile externalDependency.mockitoAll
+  testCompile externalDependency.cglib
+  testCompile externalDependency.objenesis
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+

cluster/src/main/scala/com/linkedin/norbert/cluster/zookeeper/ZooKeeperClusterManagerComponent.scala

@@ -37,6 +37,8 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
     //synchronization is not needed between this method
     case class NodeChildrenChanged(path: String) extends ZooKeeperMessage
     case class NodeDataChanged(path: String) extends ZooKeeperMessage
+    case class NodeDeleted(path: String) extends ZooKeeperMessage
+    case class NodeCreated(path: String) extends ZooKeeperMessage
   }
 
   class ZooKeeperClusterManager(connectString: String, sessionTimeout: Int, serviceName: String)
@@ -90,6 +92,19 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
             handleCapabilityMemberChanged(path)
           }
 
+          case NodeDeleted(path) => if (path.equals(MEMBERSHIP_NODE) || path.equals(AVAILABILITY_NODE)) {
+            //zookeeper data corrupted
+            log.fatal("Received a zookeeper corruption message")    
+          } else if (path.startsWith(MEMBERSHIP_NODE) || path.startsWith(AVAILABILITY_NODE)) {
+            log.info("Received an event where a node was deleted:%s".format(path))
+          } else {
+            log.error("Node deleted unexpectedly %s".format(path))
+          }
+
+          case NodeCreated(path) => {
+            log.error("Received an unexpected create event:%s".format(path))
+          }
+
           case m => log.error("Received unknown message: %s".format(m))
         }
       }
@@ -461,6 +476,10 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
         case EventType.NodeChildrenChanged => zooKeeperManager ! NodeChildrenChanged(event.getPath)
 
         case EventType.NodeDataChanged => zooKeeperManager ! NodeDataChanged(event.getPath)
+
+        case EventType.NodeCreated => zooKeeperManager ! NodeCreated(event.getPath)
+
+        case EventType.NodeDeleted => zooKeeperManager ! NodeDeleted(event.getPath)
       }
     }
 

cluster/src/main/scala/com/linkedin/norbert/jmx/AverageTimeTracker.scala

@@ -71,23 +71,131 @@ class FinishedRequestTimeTracker(clock: Clock, interval: Long) {
   }
 }
 
+class TotalRequestProcessingTime[KeyT](clock:Clock, interval:Long) {
+  private val q = new java.util.concurrent.ConcurrentLinkedQueue[(Long, Long)]()
+  private val currentlyCleaning = new java.util.concurrent.atomic.AtomicBoolean
+
+  private def clean {
+    // Let only one thread clean at a time
+    if(currentlyCleaning.compareAndSet(false, true)) {
+      clean0
+      currentlyCleaning.set(false)
+    }
+  }
+
+  private def clean0 {
+    while(!q.isEmpty) {
+      val head = q.peek
+      if(head == null)
+        return
+
+      val (completion, processingTime) = head
+      if(clock.getCurrentTimeOffsetMicroseconds - completion > interval) {
+        q.remove(head)
+      } else {
+        return
+      }
+    }
+  }
+
+  def addTime(processingTime: Long) {
+    clean
+    q.offer( (clock.getCurrentTimeOffsetMicroseconds, processingTime) )
+  }
+
+  def getArray: Array[(Long, Long)] = {
+    clean
+    q.toArray(Array.empty[(Long, Long)])
+  }
+
+  def getTimings: Array[Long] = {
+    getArray.map(_._2).sorted
+  }
+
+  def total = {
+    getTimings.sum
+  }
+
+  def reset {
+    q.clear
+  }
+}
+
+class QueueTimeTracker[KeyT](clock: Clock, interval: Long) {
+    private val q = new java.util.concurrent.ConcurrentLinkedQueue[(Long, Long)]()
+    private val currentlyCleaning = new java.util.concurrent.atomic.AtomicBoolean
+
+    private def clean {
+      // Let only one thread clean at a time
+      if(currentlyCleaning.compareAndSet(false, true)) {
+        clean0
+        currentlyCleaning.set(false)
+      }
+    }
+
+    private def clean0 {
+      while(!q.isEmpty) {
+        val head = q.peek
+        if(head == null)
+          return
+
+        val (completion, processingTime) = head
+        if(clock.getCurrentTimeOffsetMicroseconds - completion > interval) {
+          q.remove(head)
+        } else {
+          return
+        }
+      }
+    }
+
+    def addTime(processingTime: Long) {
+      clean
+      q.offer( (clock.getCurrentTimeOffsetMicroseconds, processingTime) )
+    }
+
+    def getArray: Array[(Long, Long)] = {
+      clean
+      q.toArray(Array.empty[(Long, Long)])
+    }
+
+    def getTimings: Array[Long] = {
+      getArray.map(_._2).sorted
+    }
+
+    def total = {
+      getTimings.sum
+    }
+
+    def reset {
+      q.clear
+    }
+}
+
 // Threadsafe
 class PendingRequestTimeTracker[KeyT](clock: Clock) {
   private val numRequests = new AtomicInteger()
 
   private val map : java.util.concurrent.ConcurrentMap[KeyT, Long] =
     new java.util.concurrent.ConcurrentHashMap[KeyT, Long]
 
+  private val mapQueueTime : java.util.concurrent.ConcurrentMap[KeyT, Long] = 
+    new java.util.concurrent.ConcurrentHashMap[KeyT, Long]
+
   def getStartTime(key: KeyT) = Option(map.get(key))
 
-  def beginRequest(key: KeyT) {
+  //pre-condition for this method is the above method returns some 
+  def getQueueTime(key: KeyT) = map.get(key)
+
+  def beginRequest(key: KeyT, queueTime: Long) {
     numRequests.incrementAndGet
     val now = clock.getCurrentTimeOffsetMicroseconds
     map.put(key, now)
+    mapQueueTime.put(key, queueTime)
   }
 
   def endRequest(key: KeyT) {
     map.remove(key)
+    mapQueueTime.remove(key)
   }
 
   def getTimings = {
@@ -108,20 +216,29 @@ class PendingRequestTimeTracker[KeyT](clock: Clock) {
 class RequestTimeTracker[KeyT](clock: Clock, interval: Long) {
   val finishedRequestTimeTracker = new FinishedRequestTimeTracker(clock, interval)
   val pendingRequestTimeTracker = new PendingRequestTimeTracker[KeyT](clock)
+  val queueTimeTracker = new QueueTimeTracker[KeyT](clock, interval)//TODO
+  val totalRequestProcessingTimeTracker = new TotalRequestProcessingTime[KeyT](clock, interval)
 
-  def beginRequest(key: KeyT) {
-    pendingRequestTimeTracker.beginRequest(key)
+  def beginRequest(key: KeyT, queueTime: Long = 0) {
+    pendingRequestTimeTracker.beginRequest(key, queueTime)
   }
 
   def endRequest(key: KeyT) {
     pendingRequestTimeTracker.getStartTime(key).foreach { startTime =>
+      //over time we will retire this since this does not account for the amount of time the request
+      //was stuck in the queue
       finishedRequestTimeTracker.addTime(clock.getCurrentTimeOffsetMicroseconds - startTime)
+      val queueTime = pendingRequestTimeTracker.getQueueTime(key)
+      queueTimeTracker.addTime(queueTime)
+      totalRequestProcessingTimeTracker.addTime(queueTime + clock.getCurrentTimeOffsetMicroseconds - startTime)
     }
     pendingRequestTimeTracker.endRequest(key)
   }
 
   def reset {
     finishedRequestTimeTracker.reset
     pendingRequestTimeTracker.reset
+    queueTimeTracker.reset
+    totalRequestProcessingTimeTracker.reset
   }
 }

cluster/src/test/scala/com/linkedin/norbert/jmx/AverageTimeTrackerSpec.scala

@@ -40,10 +40,10 @@ class AverageTimeTrackerSpec extends Specification {
 
        (0 until 10).foreach { i =>
          MockClock.currentTime = 1000L * i
-         tracker.beginRequest(i)
+         tracker.beginRequest(i,0)
          (tracker.total / (i + 1)) must be_==(1000L * i / 2)
        }
     }
 
   }
-}
+}

defaultEnvironment.gradle

@@ -0,0 +1,6 @@
+subprojects {
+  repositories {
+    mavenCentral()
+  }
+}
+

examples/src/main/resources/log4j.properties

@@ -3,11 +3,11 @@ log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
 log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
 log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} - %-5p [%t:%C{1}@%L] - %m%n
 
-#log4j.appender.R=org.apache.log4j.RollingFileAppender
-#log4j.appender.R.File=/tmp/norbert.log
-#log4j.appender.R.MaxFileSize=100KB
-#log4j.appender.R.MaxBackupIndex=1
-#log4j.appender.R.layout=org.apache.log4j.PatternLayout
-#log4j.appender.R.layout.ConversionPattern=%p %t %c - %m%n
+log4j.appender.R=org.apache.log4j.RollingFileAppender
+log4j.appender.R.File=/tmp/norbert.log
+log4j.appender.R.MaxFileSize=100KB
+log4j.appender.R.MaxBackupIndex=1
+log4j.appender.R.layout=org.apache.log4j.PatternLayout
+log4j.appender.R.layout.ConversionPattern=%p %t %c - %m%n
 
-#log4j.category.com.linkedin.norbert=debug
+log4j.category.com.linkedin.norbert=debug

gradle.properties

@@ -0,0 +1 @@
+version=0.6.36

java-cluster/build.gradle

@@ -0,0 +1,11 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile project(':cluster')
+  compile externalDependency.scalaLibrary 
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+

java-cluster/src/main/scala/com/linkedin/norbert/javacompat.scala → java-cluster/src/main/scala/com/linkedin/norbert/javacompat/package.scala

@@ -28,16 +28,16 @@ package object javacompat {
   implicit def javaSetToImmutableSet[T](nodes: java.util.Set[T]): Set[T] = {
     collection.JavaConversions.asScalaSet(nodes).foldLeft(Set[T]()) { (set, n) => set + n }
   }
-  
+
   implicit def javaIntegerSetToScalaIntSet(set: java.util.Set[java.lang.Integer]): Set[Int] = {
     collection.JavaConversions.asScalaSet(set).foldLeft(collection.immutable.Set.empty[Int]) { _ + _.intValue }
   }
-  
+
   implicit def scalaIntSetToJavaIntegerSet(set: Set[Int]): java.util.Set[java.lang.Integer] = {
     val result = new java.util.HashSet[java.lang.Integer](set.size)
     set.foreach (result add _)
     result
-  }  
+  }
 
   implicit def scalaNodeToJavaNode(node: SNode): JNode = {
     if (node == null) null else JavaNode(node)

java-network/build.gradle

@@ -0,0 +1,12 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile project(':network')
+  compile project(':java-cluster')
+  compile externalDependency.scalaLibrary 
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+