Pass router param inputs to identifiers (linkerd#2056)

Stack params are passed to identifier plugins to configure them.  However, only params from the router config are passed to the identifier.  In particular, this means that the stats receiver is not available to the identifier because it is not a router config param.

Pass stack params as input to the router params.  This allows the stats receiver to be passed into the router params.

Fixes linkerd#1963 

Signed-off-by: Alex Leong <alex@buoyant.io>

linkerd/core/src/main/scala/io/buoyant/linkerd/Router.scala

@@ -152,7 +152,7 @@ trait RouterConfig {
   def disabled = protocol.experimentalRequired && !_experimentalEnabled.contains(true)
 
   @JsonIgnore
-  def routerParams = (Stack.Params.empty +
+  def routerParams(params: Stack.Params) = (params +
     param.ResponseClassifier(defaultResponseClassifier) +
     FailureAccrualConfig.default)
     .maybeWith(dtab.map(dtab => RoutingFactory.BaseDtab(() => dtab)))
@@ -167,7 +167,7 @@ trait RouterConfig {
 
   @JsonIgnore
   def router(params: Stack.Params): Router = {
-    val prms = params ++ routerParams
+    val prms = params ++ routerParams(params)
     val param.Label(label) = prms[param.Label]
     val announcers = _announcers.toSeq.flatten.map { announcer =>
       announcer.prefix -> announcer.mk(params)

linkerd/core/src/test/scala/io/buoyant/linkerd/RouterTest.scala

@@ -28,7 +28,7 @@ class RouterTest extends FunSuite with Exceptions {
   ): Router = {
     val mapper = Parser.objectMapper(yaml, Iterable(protos, interpreters, announcers))
     val cfg = mapper.readValue[RouterConfig](yaml)
-    val interpreter = cfg.interpreter.interpreter(cfg.routerParams)
+    val interpreter = cfg.interpreter.interpreter(cfg.routerParams(Stack.Params.empty))
     cfg.router(params + DstBindingFactory.Namer(interpreter))
   }
 
@@ -163,7 +163,7 @@ servers:
          |  trees: 999
          |  bounds: 998
          |""".stripMargin
-    val capacity = parseConfig(yaml).routerParams[DstBindingFactory.Capacity]
+    val capacity = parseConfig(yaml).routerParams(Stack.Params.empty)[DstBindingFactory.Capacity]
     assert(capacity.paths == 1000)
     assert(capacity.trees == 999)
     assert(capacity.bounds == 998)
@@ -175,7 +175,7 @@ servers:
       """|protocol: plain
          |originator: true
          |""".stripMargin
-    val originator = parseConfig(yaml).routerParams[Originator.Param]
+    val originator = parseConfig(yaml).routerParams(Stack.Params.empty)[Originator.Param]
     assert(originator.originator)
   }
 }

linkerd/core/src/test/scala/io/buoyant/linkerd/TestProtocol.scala

@@ -132,7 +132,7 @@ case class Fancy(fancy: Option[Boolean]) extends RouterConfig {
   override def protocol: ProtocolInitializer = TestProtocol.Fancy
 
   @JsonIgnore
-  override def routerParams: Params = super.routerParams
+  override def routerParams(params: Params): Params = super.routerParams(params)
     .maybeWith(fancy.map(FancyParam(_)))
 }
 

linkerd/protocol/h2/src/main/scala/io/buoyant/linkerd/protocol/H2Config.scala

@@ -120,17 +120,14 @@ case class H2Config(
   }
 
   @JsonIgnore
-  private[this] def routerParamsPartial: Stack.Params =
-    (super.routerParams + identifierParam)
+  override def routerParams(params: Stack.Params): Stack.Params =
+    (super.routerParams(params) + identifierParam)
       .maybeWith(h2AccessLog.map(H2AccessLogger.param.File.apply))
       .maybeWith(h2AccessLogRollPolicy.map(Policy.parse _ andThen H2AccessLogger.param.RollPolicy.apply))
       .maybeWith(h2AccessLogAppend.map(H2AccessLogger.param.Append.apply))
       .maybeWith(h2AccessLogRotateCount.map(H2AccessLogger.param.RotateCount.apply))
       .maybeWith(loggerParam)
-
-  @JsonIgnore
-  override def routerParams: Stack.Params = routerParamsPartial
-    .maybeWith(tracePropagator.map(tp => H2TracePropagatorConfig.Param(tp.mk(routerParamsPartial))))
+      .maybeWith(tracePropagator.map(tp => H2TracePropagatorConfig.Param(tp.mk(params))))
 
   private[this] def identifierParam: H2.Identifier = identifier match {
     case None => h2.HeaderTokenIdentifier.param

linkerd/protocol/http/src/main/scala/io/buoyant/linkerd/protocol/HttpConfig.scala

@@ -244,7 +244,7 @@ case class HttpConfig(
   }
 
   @JsonIgnore
-  private[this] def routerParamsPartial: Stack.Params = super.routerParams
+  override def routerParams(params: Stack.Params): Stack.Params = super.routerParams(params)
     .maybeWith(httpAccessLog.map(AccessLogger.param.File.apply))
     .maybeWith(httpAccessLogRollPolicy.map(Policy.parse _ andThen AccessLogger.param.RollPolicy.apply))
     .maybeWith(httpAccessLogAppend.map(AccessLogger.param.Append.apply))
@@ -257,9 +257,6 @@ case class HttpConfig(
     .maybeWith(maxResponseKB.map(kb => hparam.MaxResponseSize(kb.kilobytes)))
     .maybeWith(streamingEnabled.map(hparam.Streaming(_)))
     .maybeWith(compressionLevel.map(hparam.CompressionLevel(_)))
-
-  @JsonIgnore
-  override def routerParams = routerParamsPartial
-    .maybeWith(combinedIdentifier(routerParamsPartial))
-    .maybeWith(tracePropagator.map(tp => HttpTracePropagatorConfig.Param(tp.mk(routerParamsPartial))))
+    .maybeWith(combinedIdentifier(params))
+    .maybeWith(tracePropagator.map(tp => HttpTracePropagatorConfig.Param(tp.mk(params))))
 }

linkerd/protocol/http/src/test/scala/io/buoyant/linkerd/protocol/http/HttpConfigTest.scala

@@ -1,12 +1,14 @@
 package io.buoyant.linkerd.protocol.http
 
+import com.twitter.finagle.buoyant.Dst
 import com.twitter.finagle.http.{Method, Request}
-import com.twitter.finagle.{Dtab, Path}
+import com.twitter.finagle.{Dtab, Path, Stack}
+import com.twitter.util.Future
 import io.buoyant.config.Parser
-import io.buoyant.linkerd.RouterConfig
-import io.buoyant.linkerd.protocol.{HttpConfig, HttpInitializer}
+import io.buoyant.linkerd.{IdentifierInitializer, RouterConfig}
+import io.buoyant.linkerd.protocol.{HttpConfig, HttpIdentifierConfig, HttpInitializer}
 import io.buoyant.router.Http
-import io.buoyant.router.RoutingFactory.IdentifiedRequest
+import io.buoyant.router.RoutingFactory.{IdentifiedRequest, Identifier}
 import io.buoyant.router.http.TimestampHeaderFilter
 import io.buoyant.test.Awaits
 import io.buoyant.test.FunSuite
@@ -54,7 +56,7 @@ class HttpConfigTest extends FunSuite with Awaits {
                   |- port: 5000
       """.stripMargin
     val config = parse(yaml)
-    val identifier = config.routerParams[Http.param.HttpIdentifier]
+    val identifier = config.routerParams(Stack.Params.empty)[Http.param.HttpIdentifier]
       .id(Path.read("/svc"), () => Dtab.empty)
     val req = Request(Method.Get, "/one/two/three")
     req.host = "host.com"
@@ -73,7 +75,7 @@ class HttpConfigTest extends FunSuite with Awaits {
                   |- port: 5000
       """.stripMargin
     val config = parse(yaml)
-    val identifier = config.routerParams[Http.param.HttpIdentifier]
+    val identifier = config.routerParams(Stack.Params.empty)[Http.param.HttpIdentifier]
       .id(Path.read("/svc"), () => Dtab.empty)
     val req = Request(Method.Get, "/one/two/three")
     req.host = "host.com"
@@ -93,7 +95,7 @@ class HttpConfigTest extends FunSuite with Awaits {
                   |- port: 5000
       """.stripMargin
     val config = parse(yaml)
-    val identifier = config.routerParams[Http.param.HttpIdentifier]
+    val identifier = config.routerParams(Stack.Params.empty)[Http.param.HttpIdentifier]
       .id(Path.read("/svc"), () => Dtab.empty)
     val req = Request(Method.Get, "/one/two/three")
     req.host = "host.com"
@@ -113,7 +115,7 @@ class HttpConfigTest extends FunSuite with Awaits {
                   |- port: 5000
       """.stripMargin
     val config = parse(yaml)
-    val identifier = config.routerParams[Http.param.HttpIdentifier]
+    val identifier = config.routerParams(Stack.Params.empty)[Http.param.HttpIdentifier]
       .id(Path.read("/svc"), () => Dtab.empty)
     val req = Request(Method.Get, "/one/two/three")
     assert(
@@ -122,6 +124,31 @@ class HttpConfigTest extends FunSuite with Awaits {
     )
   }
 
+  test("identifier gets router params") {
+    import TestIdentifierInitializer._
+
+    val yaml = s"""
+                  |protocol: http
+                  |identifier:
+                  |  kind: io.l5d.test
+                  |servers:
+                  |- port: 5000
+      """.stripMargin
+
+    val mapper = Parser.objectMapper(yaml, Iterable(Seq(HttpInitializer), Seq(TestIdentifierInitializer)))
+    val config = mapper.readValue[RouterConfig](yaml).asInstanceOf[HttpConfig]
+
+    // Scala doesn't automatically pick up this implicit for some reason.
+    // val params = Stack.Params.empty + Test("foo")
+    val params = Stack.Params.empty.+(Test("foo"))(param)
+    val identifier = config.routerParams(params)[Http.param.HttpIdentifier]
+      .id(Path.read("/svc"), () => Dtab.empty)
+    assert(
+      await(identifier(Request())).asInstanceOf[IdentifiedRequest[Request]].dst.path ==
+        Path.read("/foo")
+    )
+  }
+
   test("timestamp header") {
     val yaml = s"""
                   |protocol: http
@@ -139,3 +166,29 @@ class HttpConfigTest extends FunSuite with Awaits {
 
   }
 }
+
+class TestIdentifierInitializer extends IdentifierInitializer {
+  override def configClass = classOf[TestIdentifierConfig]
+  override val configId = "io.l5d.test"
+}
+object TestIdentifierInitializer extends TestIdentifierInitializer {
+  case class Test(value: String)
+  implicit val param = new Stack.Param[Test] {
+    override def default = Test("default")
+  }
+}
+
+case class TestIdentifierConfig() extends HttpIdentifierConfig {
+  import TestIdentifierInitializer._
+
+  override def newIdentifier(
+    prefix: Path,
+    baseDtab: () => Dtab,
+    routerParams: Stack.Params
+  ): Identifier[Request] = {
+    println("PARAMS")
+    routerParams.foreach(println)
+    val value = routerParams[Test].value
+    req => Future.value(new IdentifiedRequest(Dst.Path(Path.read(s"/$value")), req))
+  }
+}

linkerd/protocol/thrift/src/main/scala/io/buoyant/linkerd/protocol/ThriftInitializer.scala

@@ -2,9 +2,10 @@ package io.buoyant.linkerd
 package protocol
 
 import com.fasterxml.jackson.annotation.{JsonIgnore, JsonProperty, JsonSubTypes, JsonTypeInfo}
+import com.twitter.finagle.Stack
 import com.twitter.finagle.Stack.Params
 import com.twitter.finagle.Thrift.param
-import com.twitter.finagle.buoyant.{PathMatcher, ParamsMaybeWith}
+import com.twitter.finagle.buoyant.{ParamsMaybeWith, PathMatcher}
 import com.twitter.finagle.buoyant.linkerd.{ThriftClientPrep, ThriftServerPrep, ThriftTraceInitializer}
 import io.buoyant.config.types.ThriftProtocol
 import io.buoyant.router.Thrift
@@ -59,7 +60,7 @@ case class ThriftConfig(
   @JsonIgnore
   override def protocol = ThriftInitializer
 
-  override def routerParams = super.routerParams
+  override def routerParams(params: Stack.Params) = super.routerParams(params)
     .maybeWith(thriftMethodInDst.map(Thrift.param.MethodInDst(_)))
     .maybeWith(thriftProtocol.map(proto => param.ProtocolFactory(proto.factory)))
 }

linkerd/protocol/thriftmux/src/main/scala/io/buoyant/linkerd/protocol/ThriftMuxInitializer.scala

@@ -60,7 +60,7 @@ case class ThriftMuxConfig(
   @JsonIgnore
   override def protocol = ThriftMuxInitializer
 
-  override def routerParams = super.routerParams
+  override def routerParams(params: Stack.Params) = super.routerParams(params)
     .maybeWith(thriftMethodInDst.map(Thrift.param.MethodInDst(_)))
     .maybeWith(thriftProtocol.map(proto => param.ProtocolFactory(proto.factory)))
 }