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

util-core/src/main/scala/com/twitter/util/Throwables.scala

@@ -1,6 +1,7 @@
 package com.twitter.util
 
 import scala.annotation.tailrec
+import scala.collection.mutable
 
 object Throwables {
 
@@ -38,5 +39,15 @@ object Throwables {
 
   object RootCause {
     def unapply(e: Throwable): Option[Throwable] = Option(e.getCause)
+
+    def nested(ex: Throwable): Throwable = {
+      var rootException = ex
+      val exceptions = mutable.HashSet[Throwable]()
+      while (rootException.getCause != null && !exceptions.contains(rootException)) {
+        exceptions.add(rootException)
+        rootException = rootException.getCause
+      }
+      rootException
+    }
   }
 }

util-stats/src/main/scala/com/twitter/finagle/stats/ExceptionStatsHandler.scala

@@ -111,12 +111,18 @@ class CategorizingExceptionStatsHandler(
  *   failures/restartable/com.twitter.finagle.Failure/com.twitter.finagle.CancelledConnectionException: 1
  *   failures/restartable/com.twitter.finagle.Failure/com.twitter.finagle.CancelledConnectionException/java.net.ConnectException: 1
  *
+ * `f1` is also recorded dimensionally as:
+ *   failures{interrupted=true, restartable=true, exception=java.net.ConnectException}
+ *
  * `f2` is recorded as:
  *   failures
  *   failures/com.twitter.finagle.IndividualRequestTimeoutException
  *   sourcedfailures/myservice
  *   sourcedfailures/myservice/com.twitter.finagle.IndividualRequestTimeoutException
  *
+ * `f2` is also recorded dimensionally as:
+ *   failures{interrupted=true, source=myservice, restartable=true, exception=com.twitter.finagle.IndividualRequestTimeoutException}
+ *
  * @param mkLabel label prefix, default to 'failures'
  * @param mkFlags  extracting flags if the Throwable is a Failure
  * @param mkSource extracting source when configured
@@ -139,14 +145,32 @@ private[finagle] class MultiCategorizingExceptionStatsHandler(
       if (flags.isEmpty) Seq(Seq(parentLabel))
       else flags.toSeq.map(Seq(parentLabel, _))
 
-    val labels: Seq[Seq[String]] = mkSource(t) match {
+    val maybeSource = mkSource(t)
+    val labels: Seq[Seq[String]] = maybeSource match {
       case Some(service) => flagLabels :+ Seq(SourcedFailures, service)
       case None => flagLabels
     }
 
     val paths: Seq[Seq[String]] = statPaths(t, labels, rollup)
 
-    if (flags.nonEmpty) statsReceiver.counter(parentLabel).incr()
-    paths.foreach { path => statsReceiver.counter(path: _*).incr() }
+    val dimensionalLabels = Map("exception" -> Throwables.RootCause.nested(t).getClass.getName) ++
+      flags.map { flag => flag -> "true" } ++
+      maybeSource.map("source" -> _).toSeq
+
+    val dimensionalMetric = MetricBuilder.forCounter
+      .withName(parentLabel)
+      .withLabels(dimensionalLabels)
+      .withDimensionalSupport
+
+    statsReceiver.counter(dimensionalMetric).incr()
+    paths.foreach {
+      case Seq(comp) if comp == parentLabel =>
+      // The "flat" counter is used to build up the rollup hierarchies but since we also use it
+      // to hold the dimensional metric, we skip it when recording hierarchical metrics.
+      case path =>
+        statsReceiver
+          .counter(MetricBuilder.forCounter.withHierarchicalOnly.withName(path: _*))
+          .incr()
+    }
   }
 }

util-stats/src/main/scala/com/twitter/finagle/stats/RollupStatsReceiver.scala

@@ -9,37 +9,45 @@ package com.twitter.finagle.stats
  * - "/errors"
  * - "/errors/clientErrors"
  * - "/errors/clientErrors/java_net_ConnectException"
+ *
+ * @param self the [[StatsReceiver]] to proxy metrics creation to
+ * @param hierarchicalOnly whether non-root scopes should be created with the HierarchicalOnly metrics identity
  */
-class RollupStatsReceiver(protected val self: StatsReceiver) extends StatsReceiverProxy {
+class RollupStatsReceiver(protected val self: StatsReceiver, hierarchicalOnly: Boolean = false)
+    extends StatsReceiverProxy {
 
-  private[this] def tails[A](s: Seq[A]): Seq[Seq[A]] = {
+  /**
+   * @return Seq(metrics namespace -> whether the namespace is the "parent" scope)
+   */
+  private[this] def tails[A](s: Seq[A]): Seq[(Seq[A], Boolean)] = {
     s match {
       case s @ Seq(_) =>
-        Seq(s)
+        Seq(s -> true)
 
       case Seq(hd, tl @ _*) =>
-        Seq(Seq(hd)) ++ (tails(tl) map { t => Seq(hd) ++ t })
+        Seq(Seq(hd) -> true) ++ (tails(tl) map { case (t, _) => (Seq(hd) ++ t) -> false })
     }
   }
 
-  override def counter(metricBuilder: MetricBuilder) = new Counter {
-    private[this] val allCounters = BroadcastCounter(
-      tails(metricBuilder.name).map(n => self.counter(metricBuilder.withName(n: _*)))
-    )
+  private[this] def metrics(parent: MetricBuilder): Seq[MetricBuilder] = tails(parent.name).map {
+    case (name, isRoot) =>
+      val builder = parent.withName(name: _*)
+      if (isRoot || !hierarchicalOnly) builder else builder.withHierarchicalOnly
+  }
+
+  override def counter(metricBuilder: MetricBuilder): Counter = new Counter {
+    private[this] val allCounters = BroadcastCounter(metrics(metricBuilder).map(self.counter))
     def incr(delta: Long): Unit = allCounters.incr(delta)
     def metadata: Metadata = allCounters.metadata
   }
-  override def stat(metricBuilder: MetricBuilder) = new Stat {
-    private[this] val allStats = BroadcastStat(
-      tails(metricBuilder.name).map(n => self.stat(metricBuilder.withName(n: _*)))
-    )
+  override def stat(metricBuilder: MetricBuilder): Stat = new Stat {
+    private[this] val allStats = BroadcastStat(metrics(metricBuilder).map(self.stat))
     def add(value: Float): Unit = allStats.add(value)
     def metadata: Metadata = allStats.metadata
   }
 
-  override def addGauge(metricBuilder: MetricBuilder)(f: => Float) = new Gauge {
-    private[this] val underlying =
-      tails(metricBuilder.name).map(n => self.addGauge(metricBuilder.withName(n: _*))(f))
+  override def addGauge(metricBuilder: MetricBuilder)(f: => Float): Gauge = new Gauge {
+    private[this] val underlying = metrics(metricBuilder).map { self.addGauge(_)(f) }
     def remove(): Unit = underlying.foreach(_.remove())
     def metadata: Metadata = MultiMetadata(underlying.map(_.metadata))
   }

util-stats/src/test/scala/com/twitter/finagle/stats/MultiCategorizingExceptionStatsHandlerTest.scala

@@ -20,6 +20,16 @@ class MultiCategorizingExceptionStatsHandlerTest extends AnyFunSuite {
 
     val keys = receiver.counters.keys.map(_.mkString("/")).toSeq.sorted
 
+    val identity = receiver.schemas(Seq("clienterrors")).toMetricBuilder.get.identity
+    assert(
+      identity.labels == Map(
+        "interrupted" -> "true",
+        "restartable" -> "true",
+        "source" -> "service",
+        "exception" -> "java.lang.Exception"
+      )
+    )
+
     assert(receiver.counters(Seq("clienterrors")) == 1)
     assert(receiver.counters(Seq("clienterrors", "interrupted")) == 1)
     assert(receiver.counters(Seq("clienterrors", "restartable")) == 1)
@@ -99,6 +109,9 @@ class MultiCategorizingExceptionStatsHandlerTest extends AnyFunSuite {
 
     val keys = receiver.counters.keys.map(_.mkString("/")).toSeq.sorted
 
+    val identity = receiver.schemas(Seq("clienterrors")).toMetricBuilder.get.identity
+    assert(identity.labels == Map("source" -> "service", "exception" -> "java.lang.Exception"))
+
     assert(receiver.counters(Seq("clienterrors")) == 1)
     assert(receiver.counters(Seq("clienterrors", classOf[RuntimeException].getName)) == 1)
     assert(
@@ -142,6 +155,9 @@ class MultiCategorizingExceptionStatsHandlerTest extends AnyFunSuite {
 
     val keys = receiver.counters.keys.map(_.mkString("/")).toSeq.sorted
 
+    val identity = receiver.schemas(Seq("failures")).toMetricBuilder.get.identity
+    assert(identity.labels == Map("exception" -> "java.lang.Exception"))
+
     assert(receiver.counters.view.filterKeys(_.contains("failures")).size == 3)
     assert(receiver.counters.view.filterKeys(_.contains("sourcedfailures")).size == 0)
 

util-stats/src/test/scala/com/twitter/finagle/stats/StatsReceiverTest.scala

@@ -16,7 +16,7 @@ import scala.collection.mutable.ArrayBuffer
 
 class StatsReceiverTest extends AnyFunSuite {
 
-  test("RollupStatsReceiver counter/stats") {
+  test("RollupStatsReceiver counter/stats - full translation") {
     val mem = new InMemoryStatsReceiver
     val receiver = new RollupStatsReceiver(mem)
 
@@ -30,6 +30,36 @@ class StatsReceiverTest extends AnyFunSuite {
     assert(mem.counters(Seq("toto", "titi")) == 2)
     assert(mem.counters(Seq("toto", "titi", "tata")) == 1)
     assert(mem.counters(Seq("toto", "titi", "tutu")) == 1)
+
+    def identity(path: String*): IdentityType = mem.schemas(path).identity.identityType
+
+    assert(identity("toto") == IdentityType.NonDeterminate)
+    assert(identity("toto", "titi") == IdentityType.NonDeterminate)
+    assert(identity("toto", "titi", "tata") == IdentityType.NonDeterminate)
+    assert(identity("toto", "titi", "tutu") == IdentityType.NonDeterminate)
+  }
+
+  test("RollupStatsReceiver counter/stats - hierarchical only") {
+    val mem = new InMemoryStatsReceiver
+    val receiver = new RollupStatsReceiver(mem, hierarchicalOnly = true)
+
+    receiver.counter("toto", "titi", "tata").incr()
+    assert(mem.counters(Seq("toto")) == 1)
+    assert(mem.counters(Seq("toto", "titi")) == 1)
+    assert(mem.counters(Seq("toto", "titi", "tata")) == 1)
+
+    receiver.counter("toto", "titi", "tutu").incr()
+    assert(mem.counters(Seq("toto")) == 2)
+    assert(mem.counters(Seq("toto", "titi")) == 2)
+    assert(mem.counters(Seq("toto", "titi", "tata")) == 1)
+    assert(mem.counters(Seq("toto", "titi", "tutu")) == 1)
+
+    def identity(path: String*): IdentityType = mem.schemas(path).identity.identityType
+
+    assert(identity("toto") == IdentityType.NonDeterminate)
+    assert(identity("toto", "titi") == IdentityType.HierarchicalOnly)
+    assert(identity("toto", "titi", "tata") == IdentityType.HierarchicalOnly)
+    assert(identity("toto", "titi", "tutu") == IdentityType.HierarchicalOnly)
   }
 
   test("Broadcast Counter/Stat") {