Add jitter to Marathon namer poll interval (linkerd#171) (linkerd#1448)

* Add jitter to Marathon poll attempts (linkerd#171)

For very large clusters, if a bunch of linkerds all poll the Marathon namer at exactly the same time, the namer
can be overloaded.

I've added a slight random jitter to poll attempts, based on the formula suggested by @adelong in comments
on PR linkerd#772. The jitter amount is configurable in YAML. Note that I decided to just pass a function from
MarathonInitializer to AppIdNamer, rather than pass a Stream as @adelong recommended on linkerd#772 – this
seemed simpler & thus more elegant. Annotating the jitter function with `@inline` should prevent it from
creating a new call stack, so performance is probably about the same as using a `Stream`.

I've added a handful of tests for this: we test that the correct values are parsed from the config file,
that a respectable sample of the jittered TTLs are within the expected range, and that at least 50% of
the random values are unique.

Fixes linkerd#171

Signed-off-by: Eliza Weisman <[email protected]>

* Fix incorrect lower bound in jitter tests

I forgot to subtract the maximum jitter amount from the expected lower bound - my bad!

Signed-off-by: Eliza Weisman <[email protected]>

* Fix incorrect lower bound in jitter tests

I forgot to subtract the maximum jitter amount from the expected lower bound - my bad!

Signed-off-by: Eliza Weisman <[email protected]>

* Make jitter tests expect a more reasonable number of collisions

Signed-off-by: Eliza Weisman <[email protected]>

* Reduce the number of expected unique TTLs again

@adelong suggested this
(linkerd#1448 (comment))
and...he's probably right.

marathon/src/main/scala/io/buoyant/marathon/v2/AppIdNamer.scala

@@ -13,7 +13,7 @@ object AppIdNamer {
 class AppIdNamer(
   api: Api,
   prefix: Path,
-  ttl: Duration,
+  ttl: => Duration,
   timer: Timer = DefaultTimer
 ) extends Namer {
 

namer/marathon/src/main/scala/io/buoyant/namer/marathon/MarathonInitializer.scala

@@ -7,11 +7,13 @@ import com.twitter.finagle.tracing.NullTracer
 import com.twitter.finagle._
 import com.twitter.io.Buf
 import com.twitter.logging.Logger
-import com.twitter.util.{Return, Throw}
+import com.twitter.util.{Duration, Return, Throw}
 import io.buoyant.config.types.Port
 import io.buoyant.namer.{NamerConfig, NamerInitializer}
 import io.buoyant.marathon.v2.{Api, AppIdNamer}
+
 import scala.util.control.NoStackTrace
+import scala.util.Random
 
 /**
  * Supports namer configurations in the form:
@@ -24,6 +26,7 @@ import scala.util.control.NoStackTrace
  *   port:           80
  *   uriPrefix:      /marathon
  *   ttlMs:          5000
+ *   jitterMs:       50
  *   useHealthCheck: false
  * </pre>
  */
@@ -85,6 +88,12 @@ object MarathonConfig {
   private val DefaultHost = "marathon.mesos"
   private val DefaultPrefix = Path.read("/io.l5d.marathon")
 
+  // Default TTL (in milliseconds)
+  private val DefaultTtlMs = 5000
+
+  // Default range by which to jitter the TTL (also in milliseconds)
+  private val DefaultJitterMs = 50
+
   private case class SetHost(host: String) extends SimpleFilter[http.Request, http.Response] {
     def apply(req: http.Request, service: Service[http.Request, http.Response]) = {
       req.host = host
@@ -100,13 +109,26 @@ case class MarathonConfig(
   dst: Option[String],
   uriPrefix: Option[String],
   ttlMs: Option[Int],
+  jitterMs: Option[Int],
   useHealthCheck: Option[Boolean]
 ) extends NamerConfig {
   import MarathonConfig._
 
   @JsonIgnore
   override def defaultPrefix: Path = DefaultPrefix
 
+  @JsonIgnore
+  private[marathon] val ttl = ttlMs.getOrElse(DefaultTtlMs)
+
+  @JsonIgnore
+  private[marathon] val jitter = jitterMs.getOrElse(DefaultJitterMs)
+
+  /** @return a random TTL for a poll attempt */
+  @JsonIgnore
+  @inline
+  private[marathon] def nextTtl: Duration =
+    (ttl + (Random.nextDouble() * 2 - 1) * jitter).toInt.millis
+
   /**
    * Construct a namer.
    */
@@ -137,7 +159,6 @@ case class MarathonConfig(
     val useHealthCheck0 = useHealthCheck.getOrElse(false)
     val api = Api(service, uriPrefix0, useHealthCheck0)
 
-    val ttl = ttlMs.getOrElse(5000).millis
-    new AppIdNamer(api, prefix, ttl)
+    new AppIdNamer(api, prefix, nextTtl)
   }
 }

namer/marathon/src/test/scala/io/buoyant/namer/marathon/MarathonTest.scala

@@ -2,17 +2,33 @@ package io.buoyant.namer.marathon
 
 import com.twitter.finagle.util.LoadService
 import com.twitter.finagle.{Path, Stack}
+import com.twitter.conversions.time._
 import io.buoyant.config.Parser
 import io.buoyant.config.types.Port
 import io.buoyant.namer.{NamerConfig, NamerInitializer}
 import org.scalatest.FunSuite
 
 class MarathonTest extends FunSuite {
+  // minimum number of unique TTLs to generate randomly.
+  //
+  // we're testing this by generating 300 jittered TTL durations and asserting
+  // that it contains at least this many distinct unique values.
+  //
+  // since this is probabilistic, we probably want to be fairly conservative
+  // here...it *should* be producing around 95 unique values out of every 300,
+  // and if it's not, that might be a bad sign, but we're lowballing the
+  // expected minimum amount so that we don't see tests breaking due to
+  // something possible but astronomically improbable. ah, the perils of
+  // (pseudo)random number generation!
+  //
+  // change this value if you disagree with my rationale.
+  val MinUniqueTTLs = 2
 
   test("sanity") {
     // ensure it doesn't totally blowup
     // We use a name that resolves here
-    val _ = MarathonConfig(Some("localhost"), None, None, None, None, None).newNamer(Stack.Params.empty)
+    val _ = MarathonConfig(Some("localhost"), None, None, None, None, None,
+      None).newNamer(Stack.Params.empty)
   }
 
   test("service registration") {
@@ -27,6 +43,7 @@ class MarathonTest extends FunSuite {
                   |port:           80
                   |uriPrefix:      /marathon
                   |ttlMs:          300
+                  |jitterMs:       60
                   |useHealthCheck: false
       """.stripMargin
 
@@ -37,6 +54,90 @@ class MarathonTest extends FunSuite {
     assert(marathon.uriPrefix.contains("/marathon"))
     assert(marathon._prefix.contains(Path.read("/io.l5d.marathon")))
     assert(marathon.ttlMs.contains(300))
+    assert(marathon.jitterMs.contains(60))
     assert(!marathon.disabled)
   }
+
+  test("parse config with default jitter") {
+    val yaml = s"""
+                  |kind:           io.l5d.marathon
+                  |prefix:         /io.l5d.marathon
+                  |host:           localhost
+                  |port:           80
+                  |uriPrefix:      /marathon
+                  |ttlMs:          300
+                  |useHealthCheck: false
+      """.stripMargin
+
+    val mapper = Parser.objectMapper(yaml, Iterable(Seq(MarathonInitializer)))
+    val marathon = mapper.readValue[NamerConfig](yaml).asInstanceOf[MarathonConfig]
+    assert(marathon.host.contains("localhost"))
+    assert(marathon.port.contains(Port(80)))
+    assert(marathon.uriPrefix.contains("/marathon"))
+    assert(marathon._prefix.contains(Path.read("/io.l5d.marathon")))
+    assert(marathon.ttlMs.contains(300))
+    assert(marathon.jitterMs.isEmpty)
+    assert(!marathon.disabled)
+  }
+
+  test("jitter TTLs with the default jitter") {
+    val yaml = s"""
+                  |kind:           io.l5d.marathon
+                  |prefix:         /io.l5d.marathon
+                  |host:           localhost
+                  |port:           80
+                  |uriPrefix:      /marathon
+                  |ttlMs:          300
+                  |useHealthCheck: false
+      """.stripMargin
+
+    val mapper = Parser.objectMapper(yaml, Iterable(Seq(MarathonInitializer)))
+    val marathon = mapper.readValue[NamerConfig](yaml).asInstanceOf[MarathonConfig]
+    val jitters = Stream.continually(marathon.nextTtl)
+
+    val baseTtl = marathon.ttlMs.get.millis
+    val minTtl = baseTtl - 50.millis
+    val maxTtl = baseTtl + 50.millis
+
+    for (jitter <- jitters.take(300)) {
+      assert(jitter >= minTtl)
+      assert(jitter <= maxTtl)
+    }
+
+    // assert that we're generating a reasonable amount of distinct TTLs –
+    // see above for what constitutes "reasonable"
+    assert(jitters.take(300).toSet.size >= MinUniqueTTLs)
+
+  }
+
+  test("jitter TTLs with a set jitter") {
+    val yaml = s"""
+                  |kind:           io.l5d.marathon
+                  |prefix:         /io.l5d.marathon
+                  |host:           localhost
+                  |port:           80
+                  |uriPrefix:      /marathon
+                  |ttlMs:          300
+                  |jitterMs:       100
+                  |useHealthCheck: false
+      """.stripMargin
+
+    val mapper = Parser.objectMapper(yaml, Iterable(Seq(MarathonInitializer)))
+    val marathon = mapper.readValue[NamerConfig](yaml).asInstanceOf[MarathonConfig]
+    val jitters = Stream.continually(marathon.nextTtl)
+
+    val baseTtl = marathon.ttlMs.get.millis
+    val minTtl = baseTtl - 100.millis
+    val maxTtl = baseTtl + 100.millis
+
+    for (jitter <- jitters.take(300)) {
+      assert(jitter >= minTtl)
+      assert(jitter <= maxTtl)
+    }
+
+    // assert that we're generating a reasonable amount of distinct TTLs –
+    // see above for what constitutes "reasonable"
+    assert(jitters.take(300).toSet.size >= MinUniqueTTLs)
+
+  }
 }