twitter-service: add metric_metadata endpoint to admin server

Problem
Finagle stats are currently storing but not exposing Metric Schemas containing
useful metadata about metrics.

Solution
Create an endpoint (/admin/metric_metadata.json) which exposes this metadata in a
JSON format, and which allows for metrics to be filtered by name with a query
param "name" ala "m" in /admin/metrics.json.

Result
Users and tools can now query the new /admin/metric_metadata.json endpoint to obtain
the stored metadata about a metric.

JIRA Issues: CSL-8921

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

server/src/main/scala/com/twitter/server/Admin.scala

@@ -161,6 +161,13 @@ object Admin {
         group = Some(Grouping.Metrics),
         includeInIndex = true
       ),
+      Route(
+        path = "/admin/metric_metadata.json",
+        handler = new MetricMetadataQueryHandler(),
+        alias = "Metric Metadata",
+        group = Some(Grouping.Metrics),
+        includeInIndex = true
+      ),
       Route(
         path = Path.Clients,
         handler = new ClientRegistryHandler(Path.Clients),

server/src/main/scala/com/twitter/server/handler/MetricMetadataQueryHandler.scala

@@ -0,0 +1,101 @@
+package com.twitter.server.handler
+
+import com.twitter.finagle.Service
+import com.twitter.finagle.http.{MediaType, Request, Response, Uri}
+import com.twitter.io.Buf
+import com.twitter.server.util.HttpUtils.newResponse
+import com.twitter.server.util.{JsonConverter, MetricSchemaSource}
+import com.twitter.util.Future
+
+/**
+ * A handler which accepts metrics metadata queries via http query strings and returns
+ * json encoded metrics with the metadata contained in their MetricSchemas, as well as an indicator
+ * of whether or not counters are latched.
+ *
+ * @note   When passing an explicit histogram metric via ?name=, users must provide the raw histogram
+ *         name, no percentile (eg, .p99) appended.
+ *
+ * Example Request:
+ *   http://$HOST:$PORT/admin/metric_metadata?name=my/cool/counter&name=your/fine/gauge&name=my/only/histo
+ *
+ * Response:
+ *   {
+ *   "@version" : 1.0,
+ *   "counters_latched" : false,
+ *   "metrics" : [
+ *     {
+ *       "name" : "my/cool/counter",
+ *       "kind" : "counter",
+ *       "source" : {
+ *         "class": "finagle.stats.cool",
+ *         "category": "Server",
+ *         "process_path": "dc/role/zone/service"
+ *       },
+ *       "description" : "Counts how many cools are seen",
+ *       "unit" : "Requests",
+ *       "verbosity": "Verbosity(default)",
+ *       "key_indicator" : true
+ *     },
+ *     {
+ *       "name" : "your/fine/gauge",
+ *       "kind" : "gauge",
+ *       "source" : {
+ *         "class": "finagle.stats.your",
+ *         "category": "Client",
+ *         "process_path": "dc/your_role/zone/your_service"
+ *       },
+ *       "description" : "Measures how fine the downstream system is",
+ *       "unit" : "Percentage",
+ *       "verbosity": "Verbosity(debug)",
+ *       "key_indicator" : false
+ *     },
+ *     {
+ *       "name" : "my/only/histo",
+ *       "kind" : "histogram",
+ *       "source" : {
+ *         "class": "Unspecified",
+ *         "category": "NoRoleSpecified",
+ *         "process_path": "Unspecified"
+ *       },
+ *       "description" : "No description provided",
+ *       "unit" : "Unspecified",
+ *       "verbosity": "Verbosity(default)",
+ *       "key_indicator" : false,
+ *       "buckets" : [
+ *         0.5,
+ *         0.9,
+ *         0.99,
+ *         0.999,
+ *         0.9999
+ *       ]
+ *     }
+ *   ]}
+ */
+class MetricMetadataQueryHandler(source: MetricSchemaSource = new MetricSchemaSource)
+    extends Service[Request, Response] {
+
+  private[this] def query(keys: Iterable[String]) =
+    for (k <- keys; e <- source.getSchema(k)) yield e
+
+  def apply(req: Request): Future[Response] = {
+    val uri = Uri.fromRequest(req)
+
+    val latched = source.hasLatchedCounters
+    val keysParam = uri.params.getAll("name")
+
+    val metrics =
+      if (keysParam.isEmpty) source.schemaList
+      else query(keysParam)
+
+    newResponse(
+      contentType = MediaType.JsonUtf8,
+      content = Buf.Utf8(
+        JsonConverter.writeToString(
+          Map(
+            "@version" -> 1.0,
+            "counters_latched" -> latched,
+            "metrics" -> metrics
+          )))
+    )
+  }
+}

server/src/main/scala/com/twitter/server/util/JsonConverter.scala

@@ -13,6 +13,7 @@ object JsonConverter {
     factory.disable(JsonFactory.Feature.USE_THREAD_LOCAL_FOR_BUFFER_RECYCLING)
     val mapper = new ObjectMapper(factory)
       .registerModule(DefaultScalaModule)
+      .registerModule(MetricSchemaJsonModule)
     val printer = new DefaultPrettyPrinter
     printer.indentArraysWith(DefaultIndenter.SYSTEM_LINEFEED_INSTANCE)
 

server/src/main/scala/com/twitter/server/util/MetricSchemaJsonModule.scala

@@ -0,0 +1,60 @@
+package com.twitter.server.util
+
+import com.fasterxml.jackson.core.JsonGenerator
+import com.fasterxml.jackson.databind.SerializerProvider
+import com.fasterxml.jackson.databind.module.SimpleModule
+import com.fasterxml.jackson.databind.ser.std.StdSerializer
+import com.twitter.finagle.stats.{CounterSchema, GaugeSchema, HistogramSchema, MetricSchema}
+
+object SchemaSerializer extends StdSerializer[MetricSchema](classOf[MetricSchema]) {
+
+  /**
+   * This custom serializer is used to convert MetricSchemas to JSON for the metric_metadata
+   * endpoint.
+   */
+// The impetus for a customer serializer over case class serde is:
+//    1) The nested nature of MetricBuilder with MetricSchema means we do not have the ability to
+//       make decisions about which MetricBuilder fields to include based on MetricSchema type
+//       (ie, buckets should only be present for HistogramSchema).
+//    2) Reshaping the MetricBuilder to be the top level JSON object removes the ability to inject
+//       the "kind" field based on the MetricSchema type.
+//    3) The MetricBuilder class would need to be reworked to have a Source case class
+//       nested in it which would contain a few of the currently MetricBuilder level values.
+  def serialize(
+    metricSchema: MetricSchema,
+    jsonGenerator: JsonGenerator,
+    serializerProvider: SerializerProvider
+  ): Unit = {
+    jsonGenerator.writeStartObject()
+    jsonGenerator.writeStringField("name", metricSchema.metricBuilder.name.mkString("/"))
+    val dataType = metricSchema match {
+      case _: CounterSchema => "counter"
+      case _: GaugeSchema => "gauge"
+      case _: HistogramSchema => "histogram"
+    }
+    jsonGenerator.writeStringField("kind", dataType)
+    jsonGenerator.writeObjectFieldStart("source")
+    jsonGenerator.writeStringField(
+      "class",
+      metricSchema.metricBuilder.sourceClass.getOrElse("Unspecified"))
+    jsonGenerator.writeStringField("category", metricSchema.metricBuilder.role.toString)
+    jsonGenerator.writeStringField(
+      "process_path",
+      metricSchema.metricBuilder.processPath.getOrElse("Unspecified"))
+    jsonGenerator.writeEndObject()
+    jsonGenerator.writeStringField("description", metricSchema.metricBuilder.description)
+    jsonGenerator.writeStringField("unit", metricSchema.metricBuilder.units.toString)
+    jsonGenerator.writeStringField("verbosity", metricSchema.metricBuilder.verbosity.toString)
+    jsonGenerator.writeBooleanField("key_indicator", metricSchema.metricBuilder.keyIndicator)
+    if (metricSchema.isInstanceOf[HistogramSchema]) {
+      jsonGenerator.writeArrayFieldStart("buckets")
+      metricSchema.metricBuilder.percentiles.foreach(bucket => jsonGenerator.writeNumber(bucket))
+      jsonGenerator.writeEndArray()
+    }
+    jsonGenerator.writeEndObject()
+  }
+}
+
+object MetricSchemaJsonModule extends SimpleModule {
+  addSerializer(SchemaSerializer)
+}

server/src/main/scala/com/twitter/server/util/MetricSchemaSource.scala

@@ -0,0 +1,52 @@
+package com.twitter.server.util
+
+import com.twitter.finagle.stats.{MetricSchema, SchemaRegistry}
+import com.twitter.finagle.util.LoadService
+
+private[server] object MetricSchemaSource {
+  lazy val registry: Seq[SchemaRegistry] = LoadService[SchemaRegistry]()
+}
+
+/**
+ * A map from stats names to [[com.twitter.finagle.stats.StatEntry StatsEntries]]
+ * which allows for stale StatEntries up to `refreshInterval`.
+ */
+private[server] class MetricSchemaSource(
+  registry: Seq[SchemaRegistry] = MetricSchemaSource.registry) {
+
+  /**
+   * Indicates whether or not the MetricSource is using latched Counters.
+   * @note this relies on the fact that there is only one StatsRegistry and that it is
+   *       the finagle implementation.
+   */
+  lazy val hasLatchedCounters: Boolean = {
+    assert(registry.length > 0)
+    registry.head.hasLatchedCounters
+  }
+
+  /** Returns the entry for `key` if it exists */
+  def getSchema(key: String): Option[MetricSchema] = synchronized {
+    registry.map(_.schemas()).find(_.contains(key)).flatMap(_.get(key))
+  }
+
+  /** Returns all schemas */
+  def schemaList(): Iterable[MetricSchema] = synchronized {
+    registry
+      .foldLeft(IndexedSeq[MetricSchema]()) { (seq, r) =>
+        seq ++ r.schemas().values
+      }
+  }
+
+  /** Returns true if the map contains `key` and false otherwise. */
+  def contains(key: String): Boolean = synchronized {
+    registry.exists(_.schemas().contains(key))
+  }
+
+  /** Returns the set of stat keys. */
+  def keySet: Set[String] = synchronized {
+    registry
+      .foldLeft(Set[String]()) { (set, r) =>
+        set ++ r.schemas().keySet
+      }
+  }
+}