added a new http protocol configuration option, called maxErrResponse…

…KB, which can be used to limit the body of a http error response. This is different from the already existing maxResponseKB which limits the body of any http response, error or not. By default maxErrResponseKB has the same value as maxResponseKB which is 5MB. (#2291)

Here https://api.linkerd.io/1.6.1/linkerd/index.html#http-1-1-protocol, I've found two http config options here which can be used to truncate large headers or response bodies as requested for #2201:
maxHeadersKB | 8 | The maximum size of all headers in an HTTP message.
maxResponseKB | 5120 | The maximum size of a non-chunked HTTP response payload.

maxHeadersKB was already implemented and supported for l5d-err headers, I've done some tests with large dtabs and the l5d-err headers are truncated to the size specified by maxHeadersKB.

maxResponseKB was only parsed by HttpConfig.scala but was not implemented/supported by the rest of the Linderd code. This pull request adds support for maxResponseKB limit when constructing responses containing l5d-err header, what it does it just truncates the error response body to the limit specified by the maxResponseKB. The option name is somehow misleading, as one might expected the truncation to be done for the http response packet size instead of the http response body size, more precisely for error response body size. 

With this fix someone setting the config maxResponseKB / hparam.MaxResponseSize to a value lower than the 5120 default, will obtain truncation of the err response body size.

IMO A better approach would be to have two new config options specific to err headers and err response bodies, called maxErrHeadersKB and maxErrResponseKB. In this way users would benefit by the standard limits for headers and reponses and also have a much finer control over err headers and err responses.

Signed-off-by: dst4096 <[email protected]>

linkerd/docs/protocol-http.md

@@ -13,6 +13,7 @@ routers:
  kind: io.l5d.methodAndHost
  maxHeadersKB: 8
  maxInitialLineKB: 4
+ maxErrResponseKB: 5120
  servers:
  - port: 5000
  addForwardedHeader:
@@ -37,6 +38,7 @@ identifier | The `io.l5d.header.token` identifier | An identifier or list of ide
 streamAfterContentLengthKB | 5 | The threshold at which HTTP messages will be streamed if exceeded. You can use this to allow for sufficiently small messages to be buffered, instead of always streamed
 maxHeadersKB | 8 | The maximum size of all headers in an HTTP message.
 maxInitialLineKB | 4 | The maximum size of an initial HTTP message line.
+maxErrResponseKB | 5120 | The maximum size of a HTTP error response payload.
 compressionLevel | `-1`, automatically compresses textual content types with compression level 6 | The compression level to use (on 0-9).
 streamingEnabled | `true` | Streaming allows Linkerd to work with HTTP messages that have large (or infinite) content bodies using chunked encoding. Disabling this is highly discouraged.
 tracePropagator | `io.l5d.default` | A trace propagator. See [Http-specific trace propagator](#http-1-1-trace-propagators).

linkerd/protocol/http/src/main/scala/com/twitter/finagle/buoyant/linkerd/LinkerdHeaders.scala

@@ -5,7 +5,8 @@ import com.twitter.finagle.buoyant.{Dst => BuoyantDst}
 import com.twitter.finagle.context.{Contexts, Deadline => FDeadline}
 import com.twitter.finagle.http.{param => hparam, _}
 import com.twitter.finagle.tracing._
-import com.twitter.util.{Future, Return, Throw, Time, Try}
+import com.twitter.util.{Future, Return, StorageUnit, Throw, Time, Try}
+import com.twitter.conversions.StorageUnitOps._
 import java.net.URLEncoder
 import java.nio.charset.StandardCharsets.ISO_8859_1
 import java.util.Base64
@@ -49,6 +50,19 @@ import scala.collection.breakOut
 object Headers {
  val Prefix = "l5d-"
 
+ /**
+ * A param specific to error responses, the maximum size of a HTTP error response payload.
+ * By default it has the same value as maxResponseKB. Example:
+ * - protocol: http
+ * maxResponseKB: 5120
+ */
+ object param {
+ final case class MaxErrResponseSize(size: StorageUnit) extends AnyVal
+ implicit object MaxErrResponseSize extends Stack.Param[MaxErrResponseSize] {
+ val default = MaxErrResponseSize(5.megabytes)
+ }
+ }
+
  object Ctx {
 
  /**
@@ -65,15 +79,15 @@ object Headers {
  * HttpTraceInitializer.serverModule.
  */
  val serverModule: Stackable[ServiceFactory[Request, Response]] =
- new Stack.Module1[hparam.MaxHeaderSize, ServiceFactory[Request, Response]] {
+ new Stack.Module2[hparam.MaxHeaderSize, Headers.param.MaxErrResponseSize, ServiceFactory[Request, Response]] {
  val role = Stack.Role("ServerContext")
  val description = "Extracts linkerd context from http headers"
 
  val deadline = new Deadline.ServerFilter
- def dtab(maxHeaderSize: Int) = new Dtab.ServerFilter(maxHeaderSize)
+ def dtab(maxHeaderSize: Int, maxErrResponseSize: Int) = new Dtab.ServerFilter(maxHeaderSize, maxErrResponseSize)
 
- def make(maxHeaderSize: hparam.MaxHeaderSize, next: ServiceFactory[Request, Response]) =
- deadline.andThen(dtab(maxHeaderSize.size.bytes.toInt)).andThen(next)
+ def make(maxHeaderSize: hparam.MaxHeaderSize, maxErrResponseSize: Headers.param.MaxErrResponseSize, next: ServiceFactory[Request, Response]) =
+ deadline.andThen(dtab(maxHeaderSize.size.bytes.toInt, maxErrResponseSize.size.bytes.toInt)).andThen(next)
  }
 
  val clearServerModule: Stackable[ServiceFactory[Request, Response]] =
@@ -255,12 +269,12 @@ object Headers {
  *
  * @todo use DtabFilter.Injector once it is released.
  */
- class ServerFilter(maxHeaderSize: Int) extends SimpleFilter[Request, Response] {
+ class ServerFilter(maxHeaderSize: Int, maxErrResponseSize: Int) extends SimpleFilter[Request, Response] {
 
  def apply(req: Request, service: Service[Request, Response]) =
  get(req.headerMap) match {
  case Throw(e) =>
- Future.value(Err.respond(e.getMessage, Status.BadRequest, maxHeaderSize))
+ Future.value(Err.respond(e.getMessage, Status.BadRequest, maxHeaderSize, maxErrResponseSize))
  case Return(dtab) =>
  clear(req.headerMap)
  FDtab.local ++= dtab
@@ -473,7 +487,7 @@ object Headers {
  object Err {
  val Key = Prefix + "err"
 
- def respond(msg: String, status: Status = Status.InternalServerError, maxHeaderSize: Int): Response = {
+ def respond(msg: String, status: Status = Status.InternalServerError, maxHeaderSize: Int, maxErrResponseSize: Int): Response = {
  val rsp = Response(status)
  val header = URLEncoder.encode(msg, ISO_8859_1.toString)
  rsp.headerMap(Key) = if (header.length > maxHeaderSize) {
@@ -482,7 +496,13 @@ object Headers {
  header
  }
  rsp.contentType = MediaType.Txt
- rsp.contentString = msg
+
+ if (msg.length > maxErrResponseSize) {
+ rsp.contentString = msg.substring(0, maxErrResponseSize)
+ } else {
+ rsp.contentString = msg
+ }
+
  rsp
  }
  }

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

@@ -66,6 +66,7 @@ class HttpInitializer extends ProtocolInitializer.Simple {
  .configured(router.params[hparam.MaxInitialLineSize])
  .configured(router.params[hparam.MaxRequestSize])
  .configured(router.params[hparam.MaxResponseSize])
+ .configured(router.params[Headers.param.MaxErrResponseSize])
  .configured(router.params[hparam.Streaming])
  .configured(router.params[hparam.CompressionLevel])
  .configured(router.params[HttpTracePropagatorConfig.Param])
@@ -216,6 +217,7 @@ case class HttpConfig(
  streamAfterContentLengthKB: Option[Int],
  maxHeadersKB: Option[Int],
  maxInitialLineKB: Option[Int],
+ maxErrResponseKB: Option[Int],
  streamingEnabled: Option[Boolean],
  compressionLevel: Option[Int],
  tracePropagator: Option[HttpTracePropagatorConfig]
@@ -265,6 +267,7 @@ case class HttpConfig(
  .maybeWith(Some(streaming))
  .maybeWith(Some(hparam.MaxRequestSize(MaxReqRespSize)))
  .maybeWith(Some(hparam.MaxResponseSize(MaxReqRespSize)))
+ .maybeWith(maxErrResponseKB.map(kb => Headers.param.MaxErrResponseSize.apply(kb.kilobytes)))
  .maybeWith(compressionLevel.map(hparam.CompressionLevel(_)))
  .maybeWith(combinedIdentifier(params))
  .maybeWith(tracePropagator.map(tp => HttpTracePropagatorConfig.Param(tp.mk(params))))

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

@@ -11,7 +11,7 @@ import io.buoyant.router.RoutingFactory.ResponseException
 import io.buoyant.router.http.MaxCallDepthFilter
 import scala.util.control.NonFatal
 
-class ErrorResponder(maxHeaderSize: Int)
+class ErrorResponder(maxHeaderSize: Int, maxErrResponseSize: Int)
  extends SimpleFilter[Request, Response] {
  private[this] val log = Logger.get("ErrorResponseFilter")
 
@@ -23,15 +23,15 @@ class ErrorResponder(maxHeaderSize: Int)
  e match {
  case RoutingFactory.UnknownDst(_, _) =>
  log.debug(e, "unknown dst")
- Headers.Err.respond(e.getMessage, Status.BadRequest, maxHeaderSize)
+ Headers.Err.respond(e.getMessage, Status.BadRequest, maxHeaderSize, maxErrResponseSize)
  case ErrorResponder.HttpResponseException(rsp) =>
  rsp
  case e: RichNoBrokersAvailableException =>
- Headers.Err.respond(e.exceptionMessage(), Status.BadGateway, maxHeaderSize)
+ Headers.Err.respond(e.exceptionMessage(), Status.BadGateway, maxHeaderSize, maxErrResponseSize)
  case e: RichConnectionFailedExceptionWithPath =>
- Headers.Err.respond(e.exceptionMessage, Status.BadGateway, maxHeaderSize)
+ Headers.Err.respond(e.exceptionMessage, Status.BadGateway, maxHeaderSize, maxErrResponseSize)
  case e: MaxCallDepthFilter.MaxCallDepthExceeded =>
- Headers.Err.respond(e.getMessage, Status.BadRequest, maxHeaderSize)
+ Headers.Err.respond(e.getMessage, Status.BadRequest, maxHeaderSize, maxErrResponseSize)
  case _ =>
  val message = e.getMessage match {
  case null => e.getClass.getName
@@ -44,7 +44,7 @@ class ErrorResponder(maxHeaderSize: Int)
  log.error("service failure: %s", e)
  Status.BadGateway
  }
- val rsp = Headers.Err.respond(message, status, maxHeaderSize)
+ val rsp = Headers.Err.respond(message, status, maxHeaderSize, maxErrResponseSize)
  if (RetryableWriteException.unapply(e).isDefined) {
  Headers.Retryable.set(rsp.headerMap, retryable = true)
  }
@@ -56,12 +56,14 @@ class ErrorResponder(maxHeaderSize: Int)
 object ErrorResponder {
  val role = Stack.Role("ErrorResponder")
  val module: Stackable[ServiceFactory[Request, Response]] =
- new Stack.Module1[param.MaxHeaderSize, ServiceFactory[Request, Response]] {
+ new Stack.Module2[param.MaxHeaderSize, Headers.param.MaxErrResponseSize, ServiceFactory[Request, Response]] {
  val role = ErrorResponder.role
  val description = "Crafts HTTP responses for routing errors"
- def filter(maxHeaderSize: Int) = new ErrorResponder(maxHeaderSize)
- def make(maxHeaderSize: param.MaxHeaderSize, factory: ServiceFactory[Request, Response]) =
- filter(maxHeaderSize.size.bytes.toInt).andThen(factory)
+ def filter(maxHeaderSize: Int, maxErrResponseSize: Int) =
+ new ErrorResponder(maxHeaderSize, maxErrResponseSize)
+ def make(maxHeaderSize: param.MaxHeaderSize, maxErrResponseSize: Headers.param.MaxErrResponseSize,
+ factory: ServiceFactory[Request, Response]) =
+ filter(maxHeaderSize.size.bytes.toInt, maxErrResponseSize.size.bytes.toInt).andThen(factory)
  }
 
  case class HttpResponseException(rsp: Response) extends ResponseException {

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

@@ -39,7 +39,7 @@ object FramingFilter {
  /**
  * A filter that fails badly-framed requests.
  */
- def clientFilter(maxHeaderSize: Int) = new SimpleFilter[Request, Response] {
+ def clientFilter(maxHeaderSize: Int, maxErrResponseSize: Int) = new SimpleFilter[Request, Response] {
  // unlike the server filter, this needs its own logger, since
  // it responds to bad requests directly, rather than bubbling up
  // exceptions to an ErrorResponder
@@ -55,7 +55,7 @@ object FramingFilter {
  // if the exception was a FramingException, turn the Future into
  // a success by responding with Bad Request
  log.error("framing error in request", e)
- Future.value(Headers.Err.respond(e.reason, Status.BadRequest, maxHeaderSize))
+ Future.value(Headers.Err.respond(e.reason, Status.BadRequest, maxHeaderSize, maxErrResponseSize))
  case Throw(e) =>
  // other exceptions should be propagated
  Future.exception(e)
@@ -85,11 +85,12 @@ object FramingFilter {
  }
 
  val clientModule: Stackable[ServiceFactory[Request, Response]] =
- new Stack.Module1[hparam.MaxHeaderSize, ServiceFactory[Request, Response]] {
+ new Stack.Module2[hparam.MaxHeaderSize, Headers.param.MaxErrResponseSize, ServiceFactory[Request, Response]] {
  override val role: Stack.Role = FramingFilter.role
  override val description = "Fails badly-framed HTTP responses"
- override def make(maxHeaderSize: hparam.MaxHeaderSize, factory: ServiceFactory[Request, Response]): ServiceFactory[Request, Response] =
- clientFilter(maxHeaderSize.size.bytes.toInt).andThen(factory)
+ override def make(maxHeaderSize: hparam.MaxHeaderSize, maxErrResponseSize: Headers.param.MaxErrResponseSize,
+ factory: ServiceFactory[Request, Response]): ServiceFactory[Request, Response] =
+ clientFilter(maxHeaderSize.size.bytes.toInt, maxErrResponseSize.size.bytes.toInt).andThen(factory)
  }
 
  case class FramingException(reason: String) extends ProtocolException(reason)

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

@@ -5,6 +5,8 @@ import com.twitter.finagle.{Service, ServiceFactory, Stack, param}
 import com.twitter.finagle.http.{param => hparam}
 import com.twitter.finagle.http.{Request, Response, Status, Version}
 import com.twitter.finagle.stats.InMemoryStatsReceiver
+import com.twitter.finagle.buoyant.linkerd.DelayedRelease
+import com.twitter.finagle.buoyant.linkerd.Headers.{param => hErrParam}
 import com.twitter.io.Pipe
 import com.twitter.util.{Future, Promise, Time}
 import io.buoyant.linkerd.protocol.http.ResponseClassifiers
@@ -139,12 +141,14 @@ class HttpInitializerTest extends FunSuite with Awaits with Eventually {
  val maxInitLineSize = hparam.MaxInitialLineSize(30.kilobytes)
  val maxReqSize = hparam.MaxRequestSize(40.kilobytes)
  val maxRspSize = hparam.MaxResponseSize(50.kilobytes)
+ val maxErrRspSize = hErrParam.MaxErrResponseSize(30.kilobytes)
  val streaming = hparam.Streaming(false)
  val compression = hparam.CompressionLevel(3)
 
  val router = HttpInitializer.router
  .configured(maxHeaderSize).configured(maxInitLineSize)
  .configured(maxReqSize).configured(maxRspSize)
+ .configured(maxErrRspSize)
  .configured(streaming).configured(compression)
  .serving(HttpServerConfig(None, None, None).mk(HttpInitializer, "yolo"))
  .initialize()
@@ -154,6 +158,7 @@ class HttpInitializerTest extends FunSuite with Awaits with Eventually {
  assert(sparams[hparam.MaxInitialLineSize] == maxInitLineSize)
  assert(sparams[hparam.MaxRequestSize] == maxReqSize)
  assert(sparams[hparam.MaxResponseSize] == maxRspSize)
+ assert(sparams[hErrParam.MaxErrResponseSize] == maxErrRspSize)
  assert(sparams[hparam.Streaming] == streaming)
  assert(sparams[hparam.CompressionLevel] == compression)
  }