util-stats: Add metadata support to metrics instances

Problem

We want to construct metrics expressions, but it's challenging without having
access to the fully constructed `MetricBuilder` instance.  We don't have access
to the fully constructed instance because of the way StatsReceiver composition
works.  We only have partial context when we actually invoke the MetricBuilder,
and the rest of the context is assembled when the metric is constructed.

Solution

We should provide access to the fully constructed `MetricBuilder` instance after
the metric has been constructed.  We also introduce a `Metadata` abstraction to
represent when the metric actually represents multiple metrics behind the scenes.

JIRA Issues: CSL-10887

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

finagle-core/src/main/scala/com/twitter/finagle/stats/JavaLoggerStatsReceiver.scala

@@ -22,6 +22,7 @@ class JavaLoggerStatsReceiver(logger: Logger, timer: Timer)
       val formattedName = formatName(schema.metricBuilder.name)
       logger.log(level, s"$formattedName add $value")
     }
+    def metadata: Metadata = schema.metricBuilder
   }
 
   def counter(schema: CounterSchema): Counter = new Counter {
@@ -31,13 +32,17 @@ class JavaLoggerStatsReceiver(logger: Logger, timer: Timer)
       val formattedName = formatName(schema.metricBuilder.name)
       logger.log(level, s"$formattedName incr $delta")
     }
+    def metadata: Metadata = schema.metricBuilder
   }
 
   override def addGauge(schema: GaugeSchema)(f: => Float): Gauge = {
     registerGauge(schema, f)
 
-    // dummy gauge to make type signature happy.
-    new Gauge { def remove(): Unit = () }
+    // placeholder gauge that just supplies metadata
+    new Gauge {
+      def remove(): Unit = ()
+      def metadata: Metadata = schema.metricBuilder
+    }
   }
 
   protected[this] def registerGauge(schema: GaugeSchema, f: => Float): Unit =

finagle-core/src/main/scala/com/twitter/finagle/stats/SummarizingStatsReceiver.scala

@@ -30,18 +30,23 @@ class SummarizingStatsReceiver extends StatsReceiverWithCumulativeGauges {
   def counter(schema: CounterSchema): Counter = new Counter {
     counters.putIfAbsent(schema.metricBuilder.name, new AtomicLong(0))
     def incr(delta: Long): Unit = counters.get(schema.metricBuilder.name).getAndAdd(delta)
+    def metadata: Metadata = schema.metricBuilder
   }
 
   def stat(schema: HistogramSchema): Stat = new Stat {
     def add(value: Float): Unit = SummarizingStatsReceiver.this.synchronized {
       stats.get(schema.metricBuilder.name) += value
     }
+    def metadata: Metadata = schema.metricBuilder
   }
 
   override def addGauge(schema: GaugeSchema)(f: => Float): Gauge =
     synchronized {
       _gauges += (schema.metricBuilder.name -> (() => f))
-      new Gauge { def remove(): Unit = () }
+      new Gauge {
+        def remove(): Unit = ()
+        def metadata: Metadata = schema.metricBuilder
+      }
     }
 
   protected[this] def registerGauge(schema: GaugeSchema, f: => Float): Unit =

finagle-stats-core/src/main/scala/com/twitter/finagle/stats/Metrics.scala

@@ -54,13 +54,16 @@ object Metrics {
     metricBuilders = new ConcurrentHashMap[Int, MetricBuilder]()
   )
 
-  private class StoreCounterImpl(override val name: String) extends MetricsStore.StoreCounter {
+  private class StoreCounterImpl(override val name: String, _metadata: Metadata)
+      extends MetricsStore.StoreCounter {
     private[this] val adder = new LongAdder()
 
     val counter: Counter = new Counter {
       def incr(delta: Long): Unit = {
         adder.add(delta)
       }
+
+      def metadata: Metadata = _metadata
     }
 
     def count: Long = adder.sum()
@@ -71,7 +74,11 @@ object Metrics {
     override def read: Number = f
   }
 
-  private class StoreStatImpl(histo: MetricsHistogram, override val name: String, doLog: Boolean)
+  private class StoreStatImpl(
+    histo: MetricsHistogram,
+    override val name: String,
+    doLog: Boolean,
+    _metadata: Metadata)
       extends MetricsStore.StoreStat {
     def snapshot: Snapshot = histo.snapshot
 
@@ -82,6 +89,8 @@ object Metrics {
         val asLong = value.toLong
         histo.add(asLong)
       }
+
+      def metadata: Metadata = _metadata
     }
 
     def clear(): Unit = histo.clear()
@@ -167,7 +176,7 @@ private[finagle] class Metrics private (
     val curNameUsage = reservedNames.putIfAbsent(formatted, CounterRepr)
 
     if (curNameUsage == null || curNameUsage == CounterRepr) {
-      val next = new Metrics.StoreCounterImpl(formatted)
+      val next = new Metrics.StoreCounterImpl(formatted, schema.metricBuilder)
       val prev = countersMap.putIfAbsent(schema.metricBuilder.name, next)
 
       if (schema.metricBuilder.verbosity != Verbosity.Default)
@@ -214,7 +223,7 @@ private[finagle] class Metrics private (
         log.debug(s"$formatted's histogram implementation doesn't support details")
     }
 
-    val next = new Metrics.StoreStatImpl(histogram, formatted, doLog)
+    val next = new Metrics.StoreStatImpl(histogram, formatted, doLog, schema.metricBuilder)
     val prev = statsMap.putIfAbsent(schema.metricBuilder.name, next)
 
     if (schema.metricBuilder.verbosity != Verbosity.Default) {

finagle-stats-core/src/test/scala/com/twitter/finagle/stats/MetricsStatsReceiverTest.scala

@@ -246,6 +246,20 @@ class MetricsStatsReceiverTest extends FunSuite {
       assert(metrics2.schemas.containsKey("ccc"))
     }
 
+    test("StatsReceiver metrics expose the underlying schema" + suffix) {
+      val metrics = new Metrics()
+
+      val sr = new MetricsStatsReceiver(metrics)
+      val counter = addCounter(sr, Seq("aaa"))
+      assert(metrics.schemas.get("aaa").metricBuilder == counter.metadata)
+
+      val gauge = addGauge(sr, Seq("bbb"))(1f)
+      assert(metrics.schemas.get("bbb").metricBuilder == gauge.metadata)
+
+      val histo = addHisto(sr, Seq("ccc"))
+      assert(metrics.schemas.get("ccc").metricBuilder == histo.metadata)
+    }
+
     // scalafix:on StoreGaugesAsMemberVariables
   }