Skip to content

Commit

Permalink
twitter-server: expression in a query language like format
Browse files Browse the repository at this point in the history 
Problem

The general "expression" field has not yet been formatted but directly uses JSON
to reflect the class structure.

Solution

Make it a one-line string that uses the common arithmetic function names.

Result

Serialize expression needs more information (latched/unlatched), we cannot
fit it in a Jackson customized Serializer.
Added another field in the Expression case class.

JIRA Issues: CSL-10618

Differential Revision: https://phabricator.twitter.biz/D638561
  • Loading branch information
@yufangong
yufangong authored and jenkins committed Mar 29, 2021
1 parent d1cbf38 commit 48e427f
Show file tree
Hide file tree
Showing 3 changed files with 174 additions and 65 deletions.
90 changes: 85 additions & 5 deletions server/src/main/scala/com/twitter/server/handler/MetricExpressionHandler.scala
View file
Original file line number Diff line number Diff line change
@@ -1,22 +1,101 @@
package com.twitter.server.handler

import com.twitter.finagle.Service
import com.twitter.finagle.http.{MediaType, Request, Response}
import com.twitter.finagle.stats.metadataScopeSeparator
import com.twitter.finagle.http.{MediaType, Request, Response, Uri}
import com.twitter.finagle.stats.exp.Expression._
import com.twitter.finagle.stats.exp.{
ConstantExpression,
Expression,
FunctionExpression,
HistogramExpression,
MetricExpression
}
import com.twitter.finagle.stats.{
CounterSchema,
HistogramSchema,
MetricSchema,
StatsFormatter,
metadataScopeSeparator
}
import com.twitter.io.Buf
import com.twitter.server.handler.MetricExpressionHandler.Version
import com.twitter.server.handler.MetricExpressionHandler.{Version, translateToQuery}
import com.twitter.server.util.HttpUtils.newResponse
import com.twitter.server.util.{AdminJsonConverter, MetricSchemaSource}
import com.twitter.util.Future

object MetricExpressionHandler {
private val Version = 0.2
private val Version = 0.4
private val statsFormatter = StatsFormatter.default

/**
* Translate the [[Expression]] object to a single line string which represents generic
* query language.
* @param latched For unlatched counter, we wrap the metric within `rate`
*/
// exposed for testing
private[server] def translateToQuery(expr: Expression, latched: Boolean = false): String =
expr match {
case HistogramExpression(schema, component) => getHisto(schema, component)
case MetricExpression(schema) => getMetric(schema, latched)
case ConstantExpression(repr, _) => repr
case FunctionExpression(funcName, exprs) =>
s"$funcName(${exprs.map { expr => translateToQuery(expr, latched) }.mkString(",")})"
}

// Form a fully formatted name of the histogram with components
// the returned metric is styled the same way as admin/metrics.json
// e.g.request_latency.p9999 or request_latency.min
private def getHisto(
histoSchema: HistogramSchema,
histoComponent: Either[HistogramComponent, Double]
): String = {
val name = histoSchema.metricBuilder.name.mkString(metadataScopeSeparator())
val component = histoComponent match {
case Right(percentile) => statsFormatter.labelPercentile(percentile)
case Left(Min) => statsFormatter.labelMin
case Left(Max) => statsFormatter.labelMax
case Left(Avg) => statsFormatter.labelAverage
case Left(Sum) => statsFormatter.labelSum
case Left(Count) => statsFormatter.labelCount
}
statsFormatter.histoName(name, component)
}

// Form metrics other than histograms, rate() for unlatched counters
private def getMetric(
metricSchema: MetricSchema,
latched: Boolean,
): String = {
metricSchema match {
case CounterSchema(metricBuilder) if !latched =>
s"rate(${metricBuilder.name.mkString(metadataScopeSeparator())})"
case other => other.metricBuilder.name.mkString(metadataScopeSeparator())
}
}
}

/**
* A handler for metric expression queries at admin/metric/expressions.json.
* @queryParam ?latching_style=boolean Set true to let expression respect the latchedness of counters,
* which means it does not wrap the latched counters in `rate()`.
*
* @example http://$HOST:$PORT/admin/metric/expressions.json?latching_style=true
* http://$HOST:$PORT/admin/metric/expressions.json (by default latching_style is false)
*/
class MetricExpressionHandler(source: MetricSchemaSource = new MetricSchemaSource)
extends Service[Request, Response] {

private[this] lazy val sourceLatched = source.hasLatchedCounters

def apply(request: Request): Future[Response] = {
val keyParam = Uri.fromRequest(request).params.getAll("latching_style")

val latched = keyParam.exists { value => value == "true" || value == "1" } && sourceLatched

val expressions = source.expressionList.map { expressionSchema =>
expressionSchema.copy(exprQuery = translateToQuery(expressionSchema.expr, latched))
}

newResponse(
contentType = MediaType.JsonUtf8,
content = Buf.Utf8(
Expand All @@ -25,9 +104,10 @@ class MetricExpressionHandler(source: MetricSchemaSource = new MetricSchemaSourc
"@version" -> Version,
"counters_latched" -> source.hasLatchedCounters,
"separator_char" -> metadataScopeSeparator(),
"expressions" -> source.expressionList
"expressions" -> expressions
))
)
)
}

}
29 changes: 2 additions & 27 deletions server/src/main/scala/com/twitter/server/util/exp/ExpressionJson.scala
View file
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
package com.twitter.server.util.exp

import com.fasterxml.jackson.core.{JsonGenerator, JsonParser}
import com.fasterxml.jackson.databind.{DeserializationContext, SerializerProvider}
import com.fasterxml.jackson.databind.deser.std.StdDeserializer
import com.fasterxml.jackson.databind.ser.std.StdSerializer
import com.fasterxml.jackson.databind.{DeserializationContext, SerializerProvider}
import com.twitter.finagle.stats.exp._
import com.twitter.finagle.stats.metadataScopeSeparator

/**
* A set of serializers, deserializers used to serve admin/metrics/expressions endpoint
Expand Down Expand Up @@ -36,38 +35,14 @@ object ExpressionJson {
gen.writeStringField("role", expressionSchema.labels.role.toString)
gen.writeEndObject()

gen.writeObjectFieldStart("expression")
writeExpression(expressionSchema.expr, gen)
gen.writeEndObject()
gen.writeStringField("expression", expressionSchema.exprQuery)

provider.defaultSerializeField("bounds", expressionSchema.bounds, gen)

gen.writeStringField("description", expressionSchema.description)
gen.writeStringField("unit", expressionSchema.unit.toString)
gen.writeEndObject()
}

// Temporary, this is the most customized ser/deserialization in expression endpoint.
// Revisit if we need the serializer for ExpressionSchema or Expression.
def writeExpression(
expr: Expression,
gen: JsonGenerator,
name: String = "metric"
): Unit = {
expr match {
case MetricExpression(schema) =>
gen.writeStringField(name, schema.metricBuilder.name.mkString(metadataScopeSeparator()))
case FunctionExpression(funcName, exprs) =>
gen.writeStringField("operator", funcName)
gen.writeObjectFieldStart("metrics")
exprs.zipWithIndex.map {
case (expr, index) => writeExpression(expr, gen, name + "-" + index.toString)
}
gen.writeEndObject()
case ConstantExpression(repr, _) =>
gen.writeStringField("constant", repr)
}
}
}

/**
Expand Down
120 changes: 87 additions & 33 deletions server/src/test/scala/com/twitter/server/handler/exp/MetricExpressionHandlerTest.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,19 +5,23 @@ import com.twitter.finagle.http.Request
import com.twitter.finagle.stats.exp.{Expression, ExpressionSchema, GreaterThan, MonotoneThresholds}
import com.twitter.finagle.stats.{
CounterSchema,
GaugeSchema,
HistogramSchema,
InMemoryStatsReceiver,
MetricBuilder,
MetricSchema,
SchemaRegistry
}
import com.twitter.server.handler.MetricExpressionHandler
import com.twitter.server.util.{JsonUtils, MetricSchemaSource}
import com.twitter.util.Await
import com.twitter.server.util.{AdminJsonConverter, JsonUtils, MetricSchemaSource}
import com.twitter.util.{Await, Awaitable, Duration}
import org.scalatest.FunSuite

class MetricExpressionHandlerTest extends FunSuite {

private[this] def await[T](awaitable: Awaitable[T], timeout: Duration = 5.second): T =
Await.result(awaitable, timeout)

val sr = new InMemoryStatsReceiver

val successMb = CounterSchema(new MetricBuilder(name = Seq("success"), statsReceiver = sr))
Expand All @@ -28,18 +32,19 @@ class MetricExpressionHandlerTest extends FunSuite {
val successRateExpression =
ExpressionSchema(
"success_rate",
Expression(successMb).divide(Expression(successMb).plus(Expression(failuresMb))))
.withBounds(MonotoneThresholds(GreaterThan, 99.5, 99.97))
Expression(100, sr).multiply(
Expression(successMb).divide(Expression(successMb).plus(Expression(failuresMb))))
).withBounds(MonotoneThresholds(GreaterThan, 99.5, 99.97))

val throughputExpression =
ExpressionSchema("throughput", Expression(successMb).plus(Expression(failuresMb)))

val latencyExpression = ExpressionSchema("latency", Expression(latencyMb))
val latencyP99 = ExpressionSchema("latency_p99", Expression(latencyMb, Right(0.99)))

val expressionSchemaMap: Map[String, ExpressionSchema] = Map(
"success_rate" -> successRateExpression,
"throughput" -> throughputExpression,
"latency" -> latencyExpression
"latency" -> latencyP99
)

val expressionRegistry = new SchemaRegistry {
Expand All @@ -48,16 +53,36 @@ class MetricExpressionHandlerTest extends FunSuite {
val expressions: Map[String, ExpressionSchema] = expressionSchemaMap
}
val expressionSource = new MetricSchemaSource(Seq(expressionRegistry))

val expressionHandler = new MetricExpressionHandler(expressionSource)

val latchedRegistry = new SchemaRegistry {
def hasLatchedCounters: Boolean = true
def schemas(): Map[String, MetricSchema] = Map.empty
def expressions(): Map[String, ExpressionSchema] = expressionSchemaMap
}
val latchedSource = new MetricSchemaSource(Seq(latchedRegistry))
val latchedHandler = new MetricExpressionHandler(latchedSource)

val testRequest = Request("http://$HOST:$PORT/admin/metric/expressions.json")
val latchedStyleRequest = Request(
"http://$HOST:$PORT/admin/metric/expressions.json?latching_style=true")

private def getSucessRateExpression(json: String): String = {
val expressions =
AdminJsonConverter
.parse[Map[String, Any]](json).get("expressions").get.asInstanceOf[List[
Map[String, String]
]]

expressions
.filter(m => m.getOrElse("name", "") == "success_rate").head.getOrElse("expression", "")
}

test("Get the all expressions") {
val responseString =
val expectedResponse =
"""
|{
| "@version" : 0.2,
| "@version" : 0.4,
| "counters_latched" : false,
| "separator_char" : "/",
| "expressions" : [
Expand All @@ -68,17 +93,7 @@ class MetricExpressionHandlerTest extends FunSuite {
| "service_name" : "Unspecified",
| "role" : "NoRoleSpecified"
| },
| "expression" : {
| "operator" : "divide",
| "metrics" : {
| "metric-0" : "success",
| "operator" : "plus",
| "metrics" : {
| "metric-1-0" : "success",
| "metric-1-1" : "failures"
| }
| }
| },
| "expression" : "multiply(100.0,divide(rate(success),plus(rate(success),rate(failures))))",
| "bounds" : {
| "kind" : "monotone",
| "operator" : ">",
Expand All @@ -97,29 +112,21 @@ class MetricExpressionHandlerTest extends FunSuite {
| "service_name" : "Unspecified",
| "role" : "NoRoleSpecified"
| },
| "expression" : {
| "operator" : "plus",
| "metrics" : {
| "metric-0" : "success",
| "metric-1" : "failures"
| }
| },
| "expression" : "plus(rate(success),rate(failures))",
| "bounds" : {
| "kind" : "unbounded"
| },
| "description" : "Unspecified",
| "unit" : "Unspecified"
| },
| {
| "name" : "latency",
| "name" : "latency_p99",
| "labels" : {
| "process_path" : "Unspecified",
| "service_name" : "Unspecified",
| "role" : "NoRoleSpecified"
| },
| "expression" : {
| "metric" : "latency"
| },
| "expression" : "latency.p99",
| "bounds" : {
| "kind" : "unbounded"
| },
Expand All @@ -130,7 +137,54 @@ class MetricExpressionHandlerTest extends FunSuite {
|}""".stripMargin

JsonUtils.assertJsonResponse(
Await.result(expressionHandler(testRequest), 5.seconds).contentString,
responseString)
await(expressionHandler(testRequest)).contentString,
expectedResponse)
}

test("Get the latched expression with ?latched_style=true") {
val responseString = await(latchedHandler(latchedStyleRequest)).contentString

assert(
getSucessRateExpression(
responseString) == "multiply(100.0,divide(success,plus(success,failures)))")
}

test("Get the latched expression without latched_style") {
val responseString = await(latchedHandler(testRequest)).contentString

assert(getSucessRateExpression(
responseString) == "multiply(100.0,divide(rate(success),plus(rate(success),rate(failures))))")
}

test("translate expressions - counters") {
val latchedResult =
MetricExpressionHandler.translateToQuery(successRateExpression.expr, latched = true)
assert(latchedResult == "multiply(100.0,divide(success,plus(success,failures)))")

val unlatchedResult =
MetricExpressionHandler.translateToQuery(successRateExpression.expr)
assert(
unlatchedResult == "multiply(100.0,divide(rate(success),plus(rate(success),rate(failures))))")
}

test("translate histogram expressions - latched does not affect result") {
val latchedResult = MetricExpressionHandler.translateToQuery(latencyP99.expr, latched = true)
val unLatchedResult = MetricExpressionHandler.translateToQuery(latencyP99.expr)
assert(latchedResult == unLatchedResult)
}

test("translate histogram expressions - components") {
val latencyMinExpr = Expression(latencyMb, Left(Expression.Min))
val min = MetricExpressionHandler.translateToQuery(latencyMinExpr)
val p99 = MetricExpressionHandler.translateToQuery(latencyP99.expr)
assert(min == "latency.min")
assert(p99 == "latency.p99")
}

test("translate expressions - gauges") {
val connMb =
GaugeSchema(new MetricBuilder(name = Seq("client", "connections"), statsReceiver = sr))
val result = MetricExpressionHandler.translateToQuery(Expression(connMb))
assert(result == "client/connections")
}
}

0 comments on commit 48e427f

Please sign in to comment.