finagle/finagle-core: Add request classification stats in ServerStatsFilter

Problem

We'd like to be able to see on the server side what fraction of requests have
been retried in some manner -- backups, requeues, or retries. This can, for example,
be useful in diagnosing retry storms.

Solution

Add these stats under /request_classification scope.

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

Author: jcrossley

CHANGELOG.rst

@@ -37,6 +37,8 @@ New Features
 * finagle-netty4: Added support for custom event loop implementations. ``PHAB_ID=D1185136``
 * finagle-core: Add low priority offload executor. ``PHAB_ID=D1189064``
 * finagle-core: A new context, `c.t.f.Retry` is set for a request if it is a retry. ``PHAB_ID=D1228127``
+* finagle-core: Add new counters under <server_label>/request_classification for retries/requeues/backups.
+  ``PHAB_ID=D1229878``
 
 
 Bug Fixes

doc/src/sphinx/metrics/Public.rst

@@ -82,6 +82,21 @@ ServerStatsFilter
   from client to server. Be aware that clock drift between hosts, stop the world
   pauses, and queue backups can contribute here. Not supported by all protocols.
 
+**request_classification/total**
+  Counter of the total number of requests, including those with responses flagged Ignorable.
+  This can be used as a denominator when calculating the fraction of requests that are retries,
+  requeues, or backups (see counters below).
+
+**request_classification/retry**
+  Counter of the number of requests received that are retries.
+
+**request_classification/requeue**
+  Counter of the number of requests received that are requeues (e.g., retried NACKs).
+
+**request_classification/backup**
+  Counter of the number of requests that are backups.
+
+
 RequestSemaphoreFilter
 <<<<<<<<<<<<<<<<<<<<<<
 

finagle-core/src/main/scala/com/twitter/finagle/context/Requeues.scala

@@ -25,6 +25,13 @@ object Requeues extends Contexts.broadcast.Key[Requeues]("com.twitter.finagle.Re
   def current: Option[Requeues] =
     Contexts.broadcast.get(Requeues)
 
+  def isRequeue: Boolean = {
+    current match {
+      case Some(requeues) if requeues.attempt > 0 => true
+      case _ => false
+    }
+  }
+
   override def marshal(requeues: Requeues): Buf = {
     val bw = BufByteWriter.fixed(4)
     bw.writeIntBE(requeues.attempt)

finagle-core/src/main/scala/com/twitter/finagle/filter/ServerStatsFilter.scala

@@ -2,8 +2,19 @@ package com.twitter.finagle.filter
 
 import com.twitter.finagle.context.Deadline
 import com.twitter.finagle.stats.StatsReceiver
-import com.twitter.finagle.{param, Service, ServiceFactory, SimpleFilter, Stack, Stackable}
-import com.twitter.util.{Time, Duration, Future, Stopwatch}
+import com.twitter.finagle.param
+import com.twitter.finagle.Service
+import com.twitter.finagle.ServiceFactory
+import com.twitter.finagle.SimpleFilter
+import com.twitter.finagle.Stack
+import com.twitter.finagle.Stackable
+import com.twitter.finagle.context.BackupRequest
+import com.twitter.finagle.context.Requeues
+import com.twitter.finagle.context.RetryContext
+import com.twitter.util.Time
+import com.twitter.util.Duration
+import com.twitter.util.Future
+import com.twitter.util.Stopwatch
 import java.util.concurrent.TimeUnit
 
 private[finagle] object ServerStatsFilter {
@@ -27,7 +38,8 @@ private[finagle] object ServerStatsFilter {
 
 /**
  * A [[com.twitter.finagle.Filter]] that records the elapsed execution
- * times of the underlying [[com.twitter.finagle.Service]].
+ * times of the underlying [[com.twitter.finagle.Service]], in addition to classifying requests
+ * as a retry, requeue, or backup.
  *
  * @note the stat does not include the time that it takes to satisfy
  *       the returned `Future`, only how long it takes for the `Service`
@@ -42,7 +54,32 @@ private[finagle] class ServerStatsFilter[Req, Rep](
   private[this] val handletime = statsReceiver.stat("handletime_us")
   private[this] val transitTimeStat = statsReceiver.stat("transit_latency_ms")
 
+  // We use a new scope here for clarity. We record "total" for two reasons:
+  // 1. This filter is separate from the StatsFilter, which records /requests after requests have
+  //    completed --  meaning /total and /requests can be different.
+  // 2. /requests *excludes* requests that have been superceded by a backup or have their response
+  //    flagged with FailureFlags.Ignorable (for accounting with success/failures). Here, we want to
+  //    have an exact measure of requests so we can see the true proportion of retries/requeues/backups.
+  private[this] val requestClassificationScope = statsReceiver.scope("request_classification")
+  private[this] val backupCounter = requestClassificationScope.counter("backup")
+  private[this] val requeueCounter = requestClassificationScope.counter("requeue")
+  private[this] val retryCounter = requestClassificationScope.counter("retry")
+  private[this] val totalCounter = requestClassificationScope.counter("total")
+
   def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = {
+    totalCounter.incr()
+    if (BackupRequest.wasInitiated) {
+      backupCounter.incr()
+    }
+
+    if (Requeues.isRequeue) {
+      requeueCounter.incr()
+    }
+
+    if (RetryContext.isRetry) {
+      retryCounter.incr()
+    }
+
     val startAt = nowNanos()
 
     Deadline.current match {

finagle-core/src/main/scala/com/twitter/finagle/server/StackServer.scala

@@ -12,7 +12,6 @@ import com.twitter.finagle.service.TimeoutFilter
 import com.twitter.finagle.stats.ServerStatsReceiver
 import com.twitter.finagle.tracing._
 import com.twitter.finagle.Stack
-import com.twitter.finagle._
 import com.twitter.jvm.Jvm
 
 object StackServer {

finagle-core/src/test/scala/com/twitter/finagle/filter/ServerStatsFilterTest.scala

@@ -2,8 +2,14 @@ package com.twitter.finagle.filter
 
 import com.twitter.finagle.Service
 import com.twitter.finagle.stats.InMemoryStatsReceiver
-import com.twitter.util.{Stopwatch, Time, Future}
+import com.twitter.util.Future
+import com.twitter.util.Stopwatch
+import com.twitter.util.Time
 import com.twitter.conversions.DurationOps._
+import com.twitter.finagle.context
+import com.twitter.finagle.context.BackupRequest
+import com.twitter.finagle.context.Contexts
+import com.twitter.finagle.context.RetryContext
 import org.scalatest.funsuite.AnyFunSuite
 
 class ServerStatsFilterTest extends AnyFunSuite {
@@ -21,4 +27,104 @@ class ServerStatsFilterTest extends AnyFunSuite {
       assert(actual == expected)
     }
   }
+
+  test("Record retry stat when request is retry") {
+    val stats = new InMemoryStatsReceiver
+    val svc = Service.mk[Unit, Unit] { _ =>
+      Future.Done
+    }
+    val filter = new ServerStatsFilter[Unit, Unit](stats)
+    filter.andThen(svc)(())
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+
+    RetryContext.withRetry {
+      filter.andThen(svc)(())
+    }
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 1)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+    assert(stats.counters(Seq("request_classification", "total")) == 2)
+  }
+
+  test("Record requeue stat when request is requeue") {
+    val stats = new InMemoryStatsReceiver
+    val svc = Service.mk[Unit, Unit] { _ =>
+      Future.Done
+    }
+    val filter = new ServerStatsFilter[Unit, Unit](stats)
+    filter.andThen(svc)(())
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+
+    Contexts.broadcast.let(context.Requeues, context.Requeues(0)) {
+      filter.andThen(svc)(())
+    }
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+
+    Contexts.broadcast.let(context.Requeues, context.Requeues(2)) {
+      filter.andThen(svc)(())
+    }
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 1)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+    assert(stats.counters(Seq("request_classification", "total")) == 3)
+  }
+
+  test("Record retry stat when request is backup") {
+    val stats = new InMemoryStatsReceiver
+    val svc = Service.mk[Unit, Unit] { _ =>
+      Future.Done
+    }
+    val filter = new ServerStatsFilter[Unit, Unit](stats)
+    filter.andThen(svc)(())
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+
+    BackupRequest.let {
+      filter.andThen(svc)(())
+    }
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 1)
+    assert(stats.counters(Seq("request_classification", "total")) == 2)
+  }
+
+  test("Record retry+requeue+backup stat when request is retry+requeue+backup") {
+    val stats = new InMemoryStatsReceiver
+    val svc = Service.mk[Unit, Unit] { _ =>
+      Future.Done
+    }
+    val filter = new ServerStatsFilter[Unit, Unit](stats)
+    filter.andThen(svc)(())
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 0)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 0)
+    assert(stats.counters(Seq("request_classification", "backup")) == 0)
+
+    Contexts.broadcast.let(context.Requeues, context.Requeues(1)) {
+      BackupRequest.let {
+        RetryContext.withRetry {
+          filter.andThen(svc)(())
+        }
+      }
+    }
+
+    assert(stats.counters(Seq("request_classification", "retry")) == 1)
+    assert(stats.counters(Seq("request_classification", "requeue")) == 1)
+    assert(stats.counters(Seq("request_classification", "backup")) == 1)
+    assert(stats.counters(Seq("request_classification", "total")) == 2)
+  }
 }