finagle-mysql: Add metrics for the cache of prepared statements

Problem
ClientDispatcher in finagle-mysql uses a cache of prepared statements to
reuse them. However there is no observability to this cache, it is hard
to tell if the size is large enough or what the hit rate is.

Solution
Add 3 metrics showing the current state of the cache:
- pstmt-cache/calls: the total number of calls
- pstmt-cache/misses: the number of misses. The hit rate would be 1 - misses/calls then
- pstmt-cache/evicted_<reason>: a collection of counters showing how many prepared statements
  were evicted and why

Differential Revision: https://phabricator.twitter.biz/D479926

Author: mbezoyan

doc/src/sphinx/metrics/Mysql.rst

@@ -15,4 +15,13 @@
   A counter of the number of opened cursors.
 
 **<label>/cursor/closed**
-  A counter of the number of closed cursors.
+  A counter of the number of closed cursors.
+
+**<label>/pstmt-cache/calls**
+  A counter of the number of requested prepared statements.
+
+**<label>/pstmt-cache/misses**
+  A counter of the number of times when a prepared statement was not the cache.
+
+**<label>/pstmt-cache/evicted_size**
+  A counter of the number of times prepared statements were evicted from the cache.

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/ClientDispatcher.scala

@@ -8,6 +8,7 @@ import com.twitter.finagle.mysql.LostSyncException.const
 import com.twitter.finagle.mysql.param.{MaxConcurrentPrepareStatements, UnsignedColumns}
 import com.twitter.finagle.mysql.transport.{MysqlBuf, MysqlBufReader, Packet}
 import com.twitter.finagle.param.Stats
+import com.twitter.finagle.stats.{Counter, LazyStatsReceiver, StatsReceiver}
 import com.twitter.finagle.transport.Transport
 import com.twitter.finagle.{Service, ServiceProxy, Stack}
 import com.twitter.util._
@@ -24,13 +25,32 @@ case class ServerError(code: Short, sqlState: String, message: String) extends E
  * the chances of leaking prepared statements and can simplify the
  * implementation of prepared statements in the presence of a connection pool.
  */
-private[mysql] class PrepareCache(svc: Service[Request, Result], cache: Caffeine[Object, Object])
+private[mysql] class PrepareCache(
+  svc: Service[Request, Result],
+  cache: Caffeine[Object, Object],
+  statsReceiver: StatsReceiver)
     extends ServiceProxy[Request, Result](svc) {
 
+  private[this] val scopedStatsReceiver = new LazyStatsReceiver(
+    statsReceiver.scope("pstmt-cache")
+  )
+
+  private[this] val evictionCounters = {
+    val counters = new Array[Counter](RemovalCause.values().length)
+    for (value <- RemovalCause.values()) {
+      counters(value.ordinal()) =
+        scopedStatsReceiver.counter(s"evicted_${value.name().toLowerCase}")
+    }
+    counters
+  }
+  private[this] val callCounter = scopedStatsReceiver.counter("calls")
+  private[this] val missCounter = scopedStatsReceiver.counter("misses")
+
   private[this] val fn = {
     val listener = new RemovalListener[Request, Future[Result]] {
       // make sure prepared futures get removed eventually
       def onRemoval(request: Request, response: Future[Result], cause: RemovalCause): Unit = {
+        evictionCounters(cause.ordinal()).incr()
         response.respond {
           case Return(r: PrepareOK) =>
             svc(CloseRequest(r.id)).unit
@@ -43,15 +63,23 @@ private[mysql] class PrepareCache(svc: Service[Request, Result], cache: Caffeine
       .removalListener(listener)
       .build[Request, Future[Result]]()
 
-    CaffeineCache.fromCache(Service.mk { req: Request => svc(req) }, underlying)
+    CaffeineCache.fromCache(
+      fn = { req: Request =>
+        missCounter.incr()
+        svc(req)
+      },
+      cache = underlying
+    )
   }
 
   /**
    * Populate cache with unique prepare requests identified by their
    * sql queries.
    */
   override def apply(req: Request): Future[Result] = req match {
-    case _: PrepareRequest => fn(req)
+    case _: PrepareRequest =>
+      callCounter.incr()
+      fn(req)
     case _ => super.apply(req)
   }
 }
@@ -67,8 +95,9 @@ private[finagle] object ClientDispatcher {
   def apply(trans: Transport[Packet, Packet], params: Stack.Params): Service[Request, Result] = {
     val maxConcurrentPrepareStatements = params[MaxConcurrentPrepareStatements].num
     new PrepareCache(
-      new ClientDispatcher(trans, params),
-      Caffeine.newBuilder().maximumSize(maxConcurrentPrepareStatements)
+      svc = new ClientDispatcher(trans, params),
+      cache = Caffeine.newBuilder().maximumSize(maxConcurrentPrepareStatements),
+      statsReceiver = params[Stats].statsReceiver
     )
   }
 
@@ -171,8 +200,9 @@ private[finagle] final class ClientDispatcher(
           (seq2, _) <- readTx(req, ok.numOfCols)
           ps <- Future.collect(seq1.map { p => const(Field(p)) })
           cs <- Future.collect(seq2.map { p => const(Field(p)) })
-        } yield ok.copy(params = ps, columns = cs)
-
+        } yield {
+          ok.copy(params = ps, columns = cs)
+        }
         result.ensure { signal.setDone() }
 
       // decode OK Result

finagle-mysql/src/test/scala/com/twitter/finagle/mysql/unit/PrepareCacheTest.scala

@@ -2,6 +2,7 @@ package com.twitter.finagle.mysql
 
 import com.github.benmanes.caffeine.cache.Caffeine
 import com.twitter.finagle.Service
+import com.twitter.finagle.stats.{InMemoryStatsReceiver, NullStatsReceiver}
 import com.twitter.util.Future
 import java.util.concurrent.Executor
 import java.util.concurrent.LinkedBlockingQueue
@@ -24,7 +25,7 @@ class PrepareCacheTest extends FunSuite {
       .maximumSize(11)
       .executor(new Executor { def execute(r: Runnable) = r.run() })
 
-    val svc = new PrepareCache(dispatcher, cache)
+    val svc = new PrepareCache(dispatcher, cache, NullStatsReceiver)
     val r0 = PrepareRequest("SELECT 0")
     svc(r0)
     svc(r0)
@@ -54,4 +55,43 @@ class PrepareCacheTest extends FunSuite {
     for (i <- 1 to 10) svc(PrepareRequest(s"SELECT $i"))
     assert(!q.isEmpty)
   }
+
+  test("emit stats") {
+
+    val q = new LinkedBlockingQueue[Request]()
+
+    val stmtId = 2
+    val dispatcher = Service.mk[Request, Result] { req: Request =>
+      q.offer(req)
+      Future.value(PrepareOK(stmtId, 1, 1, 0))
+    }
+
+    val cache = Caffeine
+      .newBuilder()
+      .maximumSize(1)
+      .executor(new Executor { def execute(r: Runnable) = r.run() })
+
+    val ist = new InMemoryStatsReceiver()
+    val svc = new PrepareCache(dispatcher, cache, ist)
+    val r0 = PrepareRequest("SELECT 0")
+    svc(r0)
+    svc(r0)
+    assert(q.poll() == r0)
+
+    assert(ist.counters(Seq("pstmt-cache", "calls")) == 2)
+    assert(ist.counters(Seq("pstmt-cache", "misses")) == 1)
+    assert(!ist.counters.contains(Seq("pstmt-cache", "evicted_size")))
+
+    val r1 = PrepareRequest("SELECT 1")
+    svc(r1)
+
+    val expectedCounters = Seq(
+      Seq("pstmt-cache", "calls") -> 3,
+      Seq("pstmt-cache", "misses") -> 2,
+      Seq("pstmt-cache", "evicted_size") -> 1
+    ).toMap
+
+    assert(expectedCounters == ist.counters)
+  }
+
 }