finagle: more integration with dimensional metrics

=== Problem
1. Finagle & Finatra exceptions are reported only as hierarchical metrics.
2. Finatra route metrics are reported only as hierarchical metrics.
3. Certain Thrift client metrics are reported only as hierarchical metrics.

=== Solution
1. Continues to add first-class dimensional metrics support to Finagle by translating some hierarchically-scoped exception stats to labelled metrics.
2. Add dimensional equivalents to  Finatra per-route metrics.
3. Report `ServicePerEndpoint` Thrift metrics dimensionally. Exceptions here are still hierarchical, for now.

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

Author: keremnc

finagle-core/src/main/scala/com/twitter/finagle/factory/StatsFactoryWrapper.scala

@@ -1,9 +1,13 @@
 package com.twitter.finagle.factory
 
 import com.twitter.finagle._
+import com.twitter.finagle.stats.RollupStatsReceiver
+import com.twitter.finagle.stats.StatsReceiver
+import com.twitter.util.Future
+import com.twitter.util.Return
+import com.twitter.util.Stopwatch
+import com.twitter.util.Throw
 import com.twitter.util.Throwables
-import com.twitter.finagle.stats.{StatsReceiver, RollupStatsReceiver}
-import com.twitter.util.{Future, Stopwatch, Return, Throw}
 
 private[finagle] object StatsFactoryWrapper {
   val role = Stack.Role("ServiceCreationStats")
@@ -22,7 +26,10 @@ private[finagle] object StatsFactoryWrapper {
         else
           new StatsFactoryWrapper(
             next,
-            new RollupStatsReceiver(statsReceiver.scope("service_creation"))
+            new RollupStatsReceiver(
+              statsReceiver.scope("service_creation"),
+              hierarchicalOnly = true
+            )
           )
       }
     }
@@ -41,7 +48,10 @@ class StatsFactoryWrapper[Req, Rep](self: ServiceFactory[Req, Rep], statsReceive
     val elapsed = Stopwatch.start()
     super.apply(conn) respond {
       case Throw(t) =>
-        failureStats.counter(Throwables.mkString(t): _*).incr()
+        failureStats
+          .label("exception", Throwables.RootCause.nested(t).getClass.getName)
+          .counter(Throwables.mkString(t): _*)
+          .incr()
         latencyStat.add(elapsed().inMilliseconds)
       case Return(_) =>
         latencyStat.add(elapsed().inMilliseconds)

finagle-http/src/main/scala/com/twitter/finagle/http/filter/StatsFilter.scala

@@ -1,8 +1,12 @@
 package com.twitter.finagle.http.filter
 
 import com.twitter.finagle._
-import com.twitter.finagle.http.{Request, Response, Status}
-import com.twitter.finagle.stats.{Counter, Stat, StatsReceiver}
+import com.twitter.finagle.http.Request
+import com.twitter.finagle.http.Response
+import com.twitter.finagle.http.Status
+import com.twitter.finagle.stats.Counter
+import com.twitter.finagle.stats.Stat
+import com.twitter.finagle.stats.StatsReceiver
 import com.twitter.util._
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicReferenceArray
@@ -77,12 +81,17 @@ class StatsFilter[REQUEST <: Request] private[filter] (stats: StatsReceiver, now
   private[this] val statusRange: AtomicReferenceArray[StatusStats] =
     new AtomicReferenceArray[StatusStats](6)
 
+  private[this] def mkStatusStats(code: String): StatusStats = StatusStats(
+    statusReceiver.hierarchicalScope(code).label("code", code).counter(),
+    timeReceiver.hierarchicalScope(code).label("code", code).stat(),
+  )
+
   private[this] def getStatusRange(code: Int): StatusStats = {
     val index = getStatusRangeIndex(code)
     val statsPair = statusRange.get(index)
     if (statsPair == null) {
       val codeRange = statusCodeRange(code)
-      val initPair = StatusStats(statusReceiver.counter(codeRange), timeReceiver.stat(codeRange))
+      val initPair = mkStatusStats(codeRange)
       statusRange.compareAndSet(index, null, initPair)
       initPair
     } else {
@@ -94,7 +103,7 @@ class StatsFilter[REQUEST <: Request] private[filter] (stats: StatsReceiver, now
   private[this] val statusCacheFn = new java.util.function.Function[Int, StatusStats] {
     def apply(t: Int): StatusStats = {
       val statusCode = t.toString
-      StatusStats(statusReceiver.counter(statusCode), timeReceiver.stat(statusCode))
+      mkStatusStats(statusCode)
     }
   }
 

finagle-http/src/test/scala/com/twitter/finagle/http/filter/StatsFilterTest.scala

@@ -3,8 +3,6 @@ package com.twitter.finagle.http.filter
 import com.twitter.finagle.Service
 import com.twitter.finagle.http.Request
 import com.twitter.finagle.http.Response
-import com.twitter.finagle.stats.MetricBuilder.CounterType
-import com.twitter.finagle.stats.MetricBuilder.HistogramType
 import com.twitter.finagle.stats.InMemoryStatsReceiver
 import com.twitter.finagle.stats.MetricBuilder
 import com.twitter.finagle.stats.NameTranslatingStatsReceiver
@@ -13,6 +11,7 @@ import com.twitter.util.Duration
 import com.twitter.util.Future
 import com.twitter.util.Stopwatch
 import com.twitter.util.Time
+import org.mockito.ArgumentMatchers.argThat
 import org.mockito.Mockito.spy
 import org.mockito.Mockito.verify
 import org.scalatest.funsuite.AnyFunSuite
@@ -56,9 +55,9 @@ class StatsFilterTest extends AnyFunSuite {
     }
 
     // Verify that the counters and stats were only created once
-    verify(receiver).counter(MetricBuilder(name = Seq("status", "404"), metricType = CounterType))
-    verify(receiver).counter(MetricBuilder(name = Seq("status", "4XX"), metricType = CounterType))
-    verify(receiver).stat(MetricBuilder(name = Seq("time", "404"), metricType = HistogramType))
-    verify(receiver).stat(MetricBuilder(name = Seq("time", "4XX"), metricType = HistogramType))
+    verify(receiver).counter(argThat((mb: MetricBuilder) => mb.name == Seq("status", "404")))
+    verify(receiver).counter(argThat((mb: MetricBuilder) => mb.name == Seq("status", "4XX")))
+    verify(receiver).stat(argThat((mb: MetricBuilder) => mb.name == Seq("time", "404")))
+    verify(receiver).stat(argThat((mb: MetricBuilder) => mb.name == Seq("time", "4XX")))
   }
 }

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

@@ -1,16 +1,25 @@
 package com.twitter.finagle.mysql
 
-import com.github.benmanes.caffeine.cache.{Caffeine, RemovalCause, RemovalListener}
+import com.github.benmanes.caffeine.cache.Caffeine
+import com.github.benmanes.caffeine.cache.RemovalCause
+import com.github.benmanes.caffeine.cache.RemovalListener
 import com.twitter.cache.caffeine.CaffeineCache
 import com.twitter.finagle.dispatch.GenSerialClientDispatcher
 import com.twitter.finagle.dispatch.ClientDispatcher.wrapWriteException
 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.mysql.param.MaxConcurrentPrepareStatements
+import com.twitter.finagle.mysql.param.UnsignedColumns
+import com.twitter.finagle.mysql.transport.MysqlBuf
+import com.twitter.finagle.mysql.transport.MysqlBufReader
+import com.twitter.finagle.mysql.transport.Packet
 import com.twitter.finagle.param.Stats
-import com.twitter.finagle.stats.{Counter, LazyStatsReceiver, StatsReceiver}
+import com.twitter.finagle.stats.Counter
+import com.twitter.finagle.stats.LazyStatsReceiver
+import com.twitter.finagle.stats.StatsReceiver
 import com.twitter.finagle.transport.Transport
-import com.twitter.finagle.{Service, ServiceProxy, Stack}
+import com.twitter.finagle.Service
+import com.twitter.finagle.ServiceProxy
+import com.twitter.finagle.Stack
 import com.twitter.util._
 
 /**
@@ -38,8 +47,10 @@ private[mysql] class PrepareCache(
   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(value.ordinal()) = scopedStatsReceiver
+        .hierarchicalScope(s"evicted_${value.name().toLowerCase}")
+        .dimensionalScope("evicted").label("cause", value.name().toLowerCase())
+        .counter()
     }
     counters
   }

finagle-thrift/src/main/scala/com/twitter/finagle/thrift/service/ThriftReqRepServicePerEndpoint.scala

@@ -3,11 +3,19 @@ package com.twitter.finagle.thrift.service
 import com.twitter.finagle.context.Contexts
 import com.twitter.finagle.service.ResponseClassifier
 import com.twitter.finagle.stats.StatsReceiver
-import com.twitter.finagle.thrift.{Headers, RichClientParam, ThriftClientRequest, ThriftMethodStats}
-import com.twitter.finagle.{Filter, Service, SimpleFilter}
+import com.twitter.finagle.thrift.Headers
+import com.twitter.finagle.thrift.RichClientParam
+import com.twitter.finagle.thrift.ThriftClientRequest
+import com.twitter.finagle.thrift.ThriftMethodStats
+import com.twitter.finagle.Filter
+import com.twitter.finagle.Service
+import com.twitter.finagle.SimpleFilter
 import com.twitter.scrooge
 import com.twitter.scrooge.ThriftMethod
-import com.twitter.util.{Future, Return, Throw, Try}
+import com.twitter.util.Future
+import com.twitter.util.Return
+import com.twitter.util.Throw
+import com.twitter.util.Try
 
 /**
  * Construct `Service[scrooge.Request[method.Args],scrooge.Response[method.SuccessType]]` interface for a [[ThriftMethod]].
@@ -99,7 +107,11 @@ object ThriftReqRepServicePerEndpoint {
     stats: StatsReceiver,
     responseClassifier: ResponseClassifier
   ): SimpleFilter[scrooge.Request[method.Args], scrooge.Response[method.SuccessType]] = {
-    val methodStats = ThriftMethodStats(stats.scope(method.serviceName).scope(method.name))
+    val methodStats = ThriftMethodStats(
+      stats
+        .hierarchicalScope(method.serviceName).hierarchicalScope(method.name)
+        .label("service", method.serviceName).label("method", method.name)
+    )
     val methodStatsResponseHandler = ThriftMethodStatsHandler(method) _
     new SimpleFilter[scrooge.Request[method.Args], scrooge.Response[method.SuccessType]] {
       def apply(

finagle-thrift/src/main/scala/com/twitter/finagle/thrift/service/ThriftServicePerEndpoint.scala

@@ -1,9 +1,13 @@
 package com.twitter.finagle.thrift.service
 
-import com.twitter.finagle.{Filter, Service, SimpleFilter}
+import com.twitter.finagle.Filter
+import com.twitter.finagle.Service
+import com.twitter.finagle.SimpleFilter
 import com.twitter.finagle.service.ResponseClassifier
 import com.twitter.finagle.stats.StatsReceiver
-import com.twitter.finagle.thrift.{RichClientParam, ThriftClientRequest, ThriftMethodStats}
+import com.twitter.finagle.thrift.RichClientParam
+import com.twitter.finagle.thrift.ThriftClientRequest
+import com.twitter.finagle.thrift.ThriftMethodStats
 import com.twitter.scrooge.ThriftMethod
 import com.twitter.util.Future
 import org.apache.thrift.TApplicationException
@@ -82,7 +86,11 @@ object ThriftServicePerEndpoint {
     stats: StatsReceiver,
     responseClassifier: ResponseClassifier
   ): SimpleFilter[method.Args, method.SuccessType] = {
-    val methodStats = ThriftMethodStats(stats.scope(method.serviceName).scope(method.name))
+    val methodStats = ThriftMethodStats(
+      stats
+        .hierarchicalScope(method.serviceName).hierarchicalScope(method.name)
+        .label("service", method.serviceName).label("method", method.name)
+    )
     val methodStatsResponseHandler = ThriftMethodStatsHandler(method) _
     new SimpleFilter[method.Args, method.SuccessType] {
       def apply(