Wip/anekhaev/performance test

.gitignore

@@ -1,10 +1,11 @@
 .classpath
 .project
 .settings
+.cache*
 target
 .idea/
 *.iml
 logs/
 dependency-reduced-pom.xml
 nbactions.xml
-nb-configuration.xml
+nb-configuration.xml

connector/pom.xml

@@ -1,5 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
     <parent>
@@ -40,6 +41,11 @@
             <version>1.10.19</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.perf4j</groupId>
+            <artifactId>perf4j</artifactId>
+            <version>0.9.16</version>
+        </dependency>
         <dependency>
             <groupId>org.hamcrest</groupId>
             <artifactId>hamcrest-all</artifactId>

connector/src/main/scala/com/basho/riak/spark/query/Query.scala

@@ -20,12 +20,14 @@ package com.basho.riak.spark.query
 import com.basho.riak.client.api.RiakClient
 import com.basho.riak.client.api.cap.Quorum
 import com.basho.riak.client.api.commands.kv.{FetchValue, MultiFetch}
-import com.basho.riak.client.core.query.{RiakObject, Location}
-import com.basho.riak.spark.rdd.{RiakConnector, ReadConf, BucketDef}
-import scala.collection.JavaConversions._
+import com.basho.riak.client.core.query.{Location, RiakObject}
+import com.basho.riak.spark.rdd.{BucketDef, ReadConf, RiakConnector}
+import org.apache.spark.Logging
+import org.apache.spark.metrics.RiakConnectorSource
+import org.perf4j.log4j.Log4JStopWatch
 
+import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
-import org.apache.spark.Logging
 
 /**
  * Generic Riak Query
@@ -45,24 +47,68 @@ trait LocationQuery[T] extends Query[T] {
   def nextLocationChunk(nextToken: Option[_], session: RiakClient): (Option[T], Iterable[Location])
 
   def nextChunk(token: Option[_], session: RiakClient): (Option[T], Iterable[ResultT]) = {
-    val r = nextLocationChunk(token, session )
-    logDebug(s"nextLocationChunk(token=$token) returns:\n  token: ${r._1}\n  locations: ${r._2}")
-
-    dataBuffer.clear()
-
-    r match {
-      case (_, Nil) =>
-        (None, Nil)
-      case (nextToken: T, locations: Iterable[Location]) =>
-        /**
-         * To be 100% sure that massive fetch doesn't lead to the connection pool starvation,
-         * fetch will be performed by the smaller chunks of keys.
-         *
-         * Ideally the chunk size should be equal to the max number of connections for the RiakNode
-         */
-        val itChunkedLocations = locations.grouped(RiakConnector.getMinConnectionsPerNode(session))
-        fetchValues(session, itChunkedLocations, dataBuffer)
-        (nextToken, dataBuffer.toList)
+    //perf4j stop watches
+    val fullChunkSw = new Log4JStopWatch()
+    val lapSw = new Log4JStopWatch()
+
+    //codahale timers
+    val fullChunkCtx = RiakConnectorSource.instance.map(_.fbr2Full.time())
+    val notFullChunkCtx = RiakConnectorSource.instance.map(_.fbr2NotFull.time())
+    val locationsFull = RiakConnectorSource.instance.map(_.fbr2LocationsFull.time())
+    val locationsNotFull = RiakConnectorSource.instance.map(_.fbr2LocationsNotFull.time())
+
+
+    try {
+      val r = nextLocationChunk(token, session)
+      if (r._2.size == readConf.fetchSize) {
+        lapSw.lap(s"fbr-two-queries.locations.full", "Getting next chunk of locations (keys)")
+        locationsFull.map(_.stop())
+      } else {
+        lapSw.lap("fbr-two-queries.locations.notFull", s"Getting next chunk of locations (keys), size = ${r._2.size}")
+        locationsNotFull.map(_.stop())
+      }
+      logDebug(s"nextLocationChunk(token=$token) returns:\n  token: ${r._1}\n  locations: ${r._2}")
+
+      dataBuffer.clear()
+
+      r match {
+        case (_, Nil) =>
+          lapSw.lap("fbr-two-queries.empty", "Chunk of locations (keys) is empty")
+          RiakConnectorSource.instance.foreach(_.fbr2EmptyChunk.mark())
+          (None, Nil)
+        case (nextToken: T, locations: Iterable[Location]) =>
+          val valuesFull = RiakConnectorSource.instance.map(_.fbr2ValuesFull.time())
+          val valuesNotFull = RiakConnectorSource.instance.map(_.fbr2ValuesNotFull.time())
+          /**
+           * To be 100% sure that massive fetch doesn't lead to the connection pool starvation,
+           * fetch will be performed by the smaller chunks of keys.
+           *
+           * Ideally the chunk size should be equal to the max number of connections for the RiakNode
+           */
+          val itChunkedLocations = locations.grouped(RiakConnector.getMinConnectionsPerNode(session))
+          fetchValues(session, itChunkedLocations, dataBuffer)
+
+          if (readConf.fetchSize == locations.size) {
+            fullChunkSw.stop(s"fbr-two-queries.full", "Entire data chunk loaded")
+            lapSw.stop(s"fbr-two-queries.values.full", s"Getting full list of values (fetchSize = ${readConf.fetchSize})")
+            valuesFull.map(_.stop())
+            fullChunkCtx.map(_.stop())
+          } else {
+            fullChunkSw.stop("fbr-two-queries.notFull", s"Not full data chunk loaded ${locations.size}")
+            lapSw.stop(s"fbr-two-queries.values.notFull", s"Less then ${readConf.fetchSize} keys returned")
+            valuesNotFull.map(_.stop())
+            notFullChunkCtx.map(_.stop())
+          }
+
+          (nextToken, dataBuffer.toList)
+      }
+    }
+    catch {
+      case e: Throwable =>
+        fullChunkSw.stop("data-chunk.error", e)
+        lapSw.stop("data-chunk.error", e)
+        RiakConnectorSource.instance.foreach(_.fbr2ErrorChunk.mark())
+        throw e
     }
   }
 
@@ -139,7 +185,7 @@ object Query{
         require(queryData.coverageEntries.isDefined)
 
         val ce = queryData.coverageEntries.get
-        require(!ce.isEmpty)
+        require(ce.nonEmpty)
 
         if(readConf.useStreamingValuesForFBRead){
           new QueryFullBucket(bucket, readConf, ce)

connector/src/main/scala/com/basho/riak/spark/query/QueryFullBucket.scala

@@ -6,7 +6,8 @@ import com.basho.riak.client.core.operations.CoveragePlanOperation.Response.Cove
 import com.basho.riak.client.core.query.{RiakObject, Location}
 import com.basho.riak.client.core.util.BinaryValue
 import com.basho.riak.spark.rdd.{BucketDef, ReadConf}
-
+import org.apache.spark.metrics.RiakConnectorSource
+import org.perf4j.log4j.Log4JStopWatch
 import scala.collection.JavaConversions._
 
 case class QueryFullBucket(bucket: BucketDef, readConf: ReadConf, coverageEntries: Iterable[CoverageEntry]) extends Query[String] {
@@ -30,7 +31,22 @@ case class QueryFullBucket(bucket: BucketDef, readConf: ReadConf, coverageEntrie
         throw new IllegalArgumentException("Wrong nextToken")
     }
 
+
+    val sw = new Log4JStopWatch()
+    val fullCtx = RiakConnectorSource.instance.map(_.fbr1Full.time())
+    val notFullCtx = RiakConnectorSource.instance.map(_.fbr1NotFull.time())
+
     val r: FullBucketRead.Response = session.execute(builder.build())
+
+    if (r.getEntries.size() == readConf.fetchSize) {
+      sw.stop("fbr-one-query.full", s"Got ${readConf.fetchSize} entities")
+      fullCtx.map(_.stop())
+    } else {
+      sw.stop("fbr-one-query.notFull", s"Got ${readConf.fetchSize} entities")
+      notFullCtx.map(_.stop())
+    }
+
+
     val data = for {
       e <- r.getEntries
       n = (e.getLocation, e.getFetchedValue.getValue(classOf[RiakObject]))

connector/src/main/scala/com/basho/riak/spark/rdd/RiakRDD.scala

@@ -19,18 +19,16 @@ package com.basho.riak.spark.rdd
 
 import com.basho.riak.client.core.query.{Location, RiakObject}
 import com.basho.riak.client.core.util.HostAndPort
-import com.basho.riak.spark.query._
-import com.basho.riak.spark.query.{DataQueryingIterator, QueryData, Query}
-import com.basho.riak.spark.rdd.partitioner.{RiakCoveragePlanBasedPartitioner, RiakLocalCoveragePartition, RiakKeysPartition, RiakKeysPartitioner}
-import com.basho.riak.spark.util.{DataConvertingIterator, CountingIterator}
+import com.basho.riak.spark.query.{DataQueryingIterator, Query, QueryData}
+import com.basho.riak.spark.rdd.partitioner.{RiakCoveragePlanBasedPartitioner, RiakKeysPartition, RiakKeysPartitioner, RiakLocalCoveragePartition}
+import com.basho.riak.spark.util.{CountingIterator, DataConvertingIterator}
 import org.apache.spark.annotation.DeveloperApi
-
-import scala.reflect.ClassTag
-import scala.language.existentials
-
 import org.apache.spark.rdd.RDD
 import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
 
+import scala.language.existentials
+import scala.reflect.ClassTag
+
 class RiakRDD[R] private[spark] (
     @transient sc: SparkContext,
     val connector: RiakConnector,

connector/src/main/scala/org/apache/spark/metrics/RiakConnectorSource.scala

@@ -0,0 +1,60 @@
+package org.apache.spark.metrics
+
+import com.codahale.metrics
+import org.apache.spark.metrics.source.Source
+import com.codahale.metrics.Timer
+import com.codahale.metrics.UniformReservoir
+
+
+
+/**
+ * @author anekhaev
+ */
+class RiakConnectorSource extends Source {
+
+  override val sourceName = "riak-connector"
+
+  override val metricRegistry = new metrics.MetricRegistry
+
+  val fbr1Full = timer("fbr-one-query.full")
+
+  val fbr1NotFull = timer("fbr-one-query.notFull")
+
+  val fbr2LocationsFull = timer("fbr-two-queries.locations.full")
+
+  val fbr2ValuesFull = timer("fbr-two-queries.values.full")
+
+  val fbr2LocationsNotFull = timer("fbr-two-queries.locations.notFull")
+
+  val fbr2ValuesNotFull = timer("fbr-two-queries.values.notFull")
+
+  val fbr2Full = timer("fbr-two-queries.full")
+
+  val fbr2NotFull = timer("fbr-two-queries.notFull")
+
+  val fbr2EmptyChunk = metricRegistry.meter("fbr-two-queries.empty")
+
+  val fbr2ErrorChunk = metricRegistry.meter("fbr-two-queries.error")
+
+  RiakConnectorSource.registerInstance(this)
+
+
+  def timer(name: String) = {
+    metricRegistry.register(name, new Timer(new UniformReservoir()))
+  }
+}
+
+
+
+
+object RiakConnectorSource {
+
+  private var inst: Option[RiakConnectorSource] = None
+
+  def registerInstance(source: RiakConnectorSource) = synchronized {
+    inst = Some(source)
+  }
+
+  def instance = inst
+
+}